diff --git a/.env.example b/.env.example index 7016ace..e403593 100644 --- a/.env.example +++ b/.env.example @@ -19,6 +19,18 @@ DATABASE_URL=sqlite:///app.db # Application Settings DEBUG=True +# Mail Configuration (for location alerts and notifications) +MAIL_SERVER=smtp.gmail.com +MAIL_PORT=587 +MAIL_USE_TLS=true +MAIL_USERNAME=your-email@gmail.com +MAIL_PASSWORD=your-app-password +MAIL_DEFAULT_SENDER=noreply@flask2fa.com + +# Location Security Settings +MAX_LOGIN_ATTEMPTS=5 +SUSPICIOUS_LOGIN_THRESHOLD_KM=100 + # Security Headers (Production only) # SESSION_COOKIE_SECURE=True # SESSION_COOKIE_HTTPONLY=True diff --git a/app/__init__.py b/app/__init__.py index f84490c..c521c7b 100644 --- a/app/__init__.py +++ b/app/__init__.py @@ -3,6 +3,7 @@ from flask_sqlalchemy import SQLAlchemy from flask_migrate import Migrate from flask_login import LoginManager from flask_wtf.csrf import CSRFProtect +from flask_mail import Mail from config import config # Initialize extensions @@ -10,6 +11,7 @@ db = SQLAlchemy() migrate = Migrate() login_manager = LoginManager() csrf = CSRFProtect() +mail = Mail() def create_app(config_name='default'): @@ -23,19 +25,22 @@ def create_app(config_name='default'): """ app = Flask(__name__) app.config.from_object(config[config_name]) - - # Initialize extensions with app + # Initialize extensions with app db.init_app(app) migrate.init_app(app, db) csrf.init_app(app) - - # Configure Flask-Login for security + mail.init_app(app) + # Configure Flask-Login for security login_manager.init_app(app) login_manager.login_view = 'auth.login' login_manager.login_message = 'Please log in to access this page.' login_manager.login_message_category = 'info' login_manager.session_protection = 'strong' # Enhanced session protection + # Initialize mail service + from app.utils.mail import mail_service + mail_service.init_app(app, mail) + @login_manager.user_loader def load_user(user_id): """ diff --git a/app/auth/routes.py b/app/auth/routes.py index 2c02104..675c9bf 100644 --- a/app/auth/routes.py +++ b/app/auth/routes.py @@ -1,10 +1,13 @@ from flask import render_template, redirect, url_for, flash, request, session from flask_login import login_user, logout_user, current_user, login_required from urllib.parse import urlparse +from datetime import datetime, timedelta from app import db from app.auth import bp from app.auth.forms import RegistrationForm, LoginForm, TwoFactorForm -from app.models import User +from app.models import User, LoginLocation, LocationApprovalToken +from app.utils.location import location_service +from app.utils.mail import mail_service import logging # Configure logging for security events @@ -155,6 +158,9 @@ def verify_otp(): if not user.is_2fa_enabled: user.enable_2fa() + # Track login location + success = track_login_location(user) + # Update last login timestamp user.last_login = db.func.current_timestamp() db.session.commit() @@ -166,11 +172,16 @@ def verify_otp(): # Complete login process login_user(user, remember=remember_me) + # Show appropriate message based on location tracking + if success: + flash('Login successful!', 'success') + else: + flash('Login successful! Check your email for location verification.', 'info') + # Log successful login logger.info(f'Successful login for user: {user.username}') - flash('Login successful!', 'success') - # Redirect to originally requested page or dashboard + # Redirect to originally requested page or dashboard next_page = request.args.get('next') if not next_page or urlparse(next_page).netloc != '': next_page = url_for('main.index') @@ -221,3 +232,162 @@ def disable_2fa(): flash('Failed to disable two-factor authentication.', 'error') return redirect(url_for('main.profile')) + + +def track_login_location(user): + """ + Track user login location and handle suspicious login detection. + + Args: + user: User object + + Returns: + bool: True if location is trusted, False if suspicious (email sent) + """ + try: + # Get current IP and location data + ip_address = location_service.get_client_ip() + user_agent = location_service.get_user_agent() + location_data = location_service.get_location_from_ip(ip_address) + + # Check if location is suspicious + is_suspicious, last_location = location_service.is_suspicious_location( + user.id, location_data + ) + + # Create login location record + login_location = LoginLocation( + user_id=user.id, + ip_address=ip_address, + country=location_data.get('country'), + region=location_data.get('region'), + city=location_data.get('city'), + latitude=location_data.get('latitude'), + longitude=location_data.get('longitude'), + user_agent=user_agent, + is_approved=not is_suspicious, + is_suspicious=is_suspicious + ) + + db.session.add(login_location) + db.session.commit() + + # Handle suspicious login + if is_suspicious: + # Create approval token + token = LocationApprovalToken.generate_token() + approval_token = LocationApprovalToken( + user_id=user.id, + location_id=login_location.id, + token=token, + expires_at=datetime.utcnow() + timedelta(hours=24) + ) + + db.session.add(approval_token) + db.session.commit() + + # Prepare location data for email + email_location_data = { + 'city': location_data.get('city'), + 'region': location_data.get('region'), + 'country': location_data.get('country'), + 'ip_address': ip_address, + 'distance_km': last_location.get('distance_km') if last_location else None + } + + # Send suspicious login alert email + mail_service.send_suspicious_login_alert( + user.email, + user.username, + email_location_data, + token + ) + + logger.warning(f"Suspicious login detected for user {user.username} from {location_data.get('city', 'Unknown')}") + return False + + logger.info(f"Trusted location login for user {user.username} from {location_data.get('city', 'Unknown')}") + return True + + except Exception as e: + logger.error(f"Error tracking login location for user {user.username}: {str(e)}") + # On error, allow login but log the issue + return True + + +@bp.route('/approve-location/') +def approve_location(token): + """ + Approve a suspicious login location via email link. + + Args: + token: Location approval token from email + """ + try: + # Find and validate token + approval_token = LocationApprovalToken.query.filter_by(token=token).first() + + if not approval_token: + flash('Invalid or expired approval link.', 'error') + return redirect(url_for('main.index')) + + if not approval_token.is_valid(): + flash('This approval link has expired or already been used.', 'error') + return redirect(url_for('main.index')) + + # Mark location as approved + location = approval_token.location + location.is_approved = True + location.is_suspicious = False + location.approved_at = datetime.utcnow() + + # Mark token as used + approval_token.is_used = True + approval_token.used_at = datetime.utcnow() + + db.session.commit() + + # Send confirmation email + user = approval_token.user + location_data = { + 'city': location.city, + 'region': location.region, + 'country': location.country + } + + mail_service.send_location_approved_notification( + user.email, + user.username, + location_data + ) + + logger.info(f"Location approved for user {user.username}: {location.location_display}") + flash('Location approved successfully! This location is now trusted.', 'success') + + except Exception as e: + logger.error(f"Error approving location with token {token}: {str(e)}") + flash('An error occurred while approving the location.', 'error') + + return redirect(url_for('main.index')) + + +@bp.route('/login-history') +@login_required +def login_history(): + """ + Display user's login history and location data. + """ + try: + # Get user's login locations ordered by most recent + locations = LoginLocation.query.filter_by( + user_id=current_user.id + ).order_by(LoginLocation.login_time.desc()).limit(50).all() + + return render_template('auth/login_history.html', + locations=locations, + title='Login History') + + except Exception as e: + logger.error(f"Error loading login history for user {current_user.username}: {str(e)}") + flash('Error loading login history.', 'error') + return redirect(url_for('main.dashboard')) diff --git a/app/models.py b/app/models.py index 0a497e0..19a3e84 100644 --- a/app/models.py +++ b/app/models.py @@ -145,3 +145,82 @@ def validate_password_hash(target, value, oldvalue, initiator): """Ensure password hash is never stored as plaintext.""" if value and not value.startswith('pbkdf2:sha256'): raise ValueError("Password must be hashed before storage") + + +class LoginLocation(db.Model): + """ + Model to track user login locations for security monitoring. + + Security features: + - IP address tracking + - Geolocation data storage + - Login attempt tracking + - Suspicious activity detection + """ + + id = db.Column(db.Integer, primary_key=True) + user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False) + ip_address = db.Column(db.String(45), nullable=False) # IPv6 support + country = db.Column(db.String(100)) + region = db.Column(db.String(100)) + city = db.Column(db.String(100)) + latitude = db.Column(db.Float) + longitude = db.Column(db.Float) + user_agent = db.Column(db.Text) + is_approved = db.Column(db.Boolean, default=False, nullable=False) + is_suspicious = db.Column(db.Boolean, default=False, nullable=False) + login_time = db.Column(db.DateTime, default=db.func.current_timestamp()) + approved_at = db.Column(db.DateTime) + + # Relationship + user = db.relationship('User', backref=db.backref('login_locations', lazy=True, + order_by='LoginLocation.login_time.desc()')) + + def __repr__(self): + return f'' + + @property + def location_display(self): + """Human-readable location string.""" + parts = [] + if self.city: + parts.append(self.city) + if self.region and self.region != self.city: + parts.append(self.region) + if self.country: + parts.append(self.country) + return ', '.join(parts) if parts else 'Unknown Location' + + @property + def coordinates(self): + """Return coordinates as tuple if available.""" + if self.latitude is not None and self.longitude is not None: + return (self.latitude, self.longitude) + return None + + +class LocationApprovalToken(db.Model): + """ + Model for secure location approval tokens sent via email. + + Security features: + - Cryptographically secure token generation + - Token expiration + - One-time use tokens + """ + + id = db.Column(db.Integer, primary_key=True) + user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False) + location_id = db.Column(db.Integer, db.ForeignKey('login_location.id'), nullable=False) + token = db.Column(db.String(255), nullable=False, unique=True) + created_at = db.Column(db.DateTime, default=db.func.current_timestamp()) + expires_at = db.Column(db.DateTime, nullable=False) + used_at = db.Column(db.DateTime) + is_used = db.Column(db.Boolean, default=False, nullable=False) + + # Relationships + user = db.relationship('User', backref=db.backref('location_tokens', lazy=True)) + location = db.relationship('LoginLocation', backref=db.backref('approval_tokens', lazy=True)) + + def __repr__(self): + return f'' diff --git a/app/templates/auth/login_history.html b/app/templates/auth/login_history.html new file mode 100644 index 0000000..3c2f6c1 --- /dev/null +++ b/app/templates/auth/login_history.html @@ -0,0 +1,180 @@ +{% extends "base.html" %} + +{% block content %} +
+
+
+
+

+ Login History +

+
+
+

+ Review your recent login activity. If you see any suspicious activity, + please change your password immediately. +

+ + {% if locations %} +
+ + + + + + + + + + + + {% for location in locations %} + + + + + + + + {% endfor %} + +
Date & Time Location IP Address Device Status
+ {{ location.login_time.strftime('%Y-%m-%d') }}
+ {{ location.login_time.strftime('%H:%M:%S UTC') }} + 		+
+ + {{ location.location_display }} +
+ {% if location.coordinates %} + + {{ "%.4f"|format(location.latitude) }}, {{ "%.4f"|format(location.longitude) }} + + {% endif %} + 		+ {{ location.ip_address }} + + {% if location.user_agent %} + + {% if 'Chrome' in location.user_agent %} + Chrome + {% elif 'Firefox' in location.user_agent %} + Firefox + {% elif 'Safari' in location.user_agent %} + Safari + {% elif 'Edge' in location.user_agent %} + Edge + {% else %} + Unknown + {% endif %} + + {% if 'Mobile' in location.user_agent or 'Android' in location.user_agent or 'iPhone' in location.user_agent %} + + {% else %} + + {% endif %} + + {% else %} + Unknown Device + {% endif %} + + {% if location.is_approved %} + + Trusted + + {% if location.approved_at %} +
+ Approved: {{ location.approved_at.strftime('%Y-%m-%d %H:%M') }} + + {% endif %} + {% elif location.is_suspicious %} + + Suspicious + +
Pending approval + {% else %} + + Unknown + + {% endif %} +
+
+ + + {% if locations|length == 50 %} +
+ + Showing the most recent 50 login attempts. Older entries are automatically archived. +
+ {% endif %} + + {% else %} +
+ +
No Login History
+

This appears to be your first login, or login tracking was recently enabled.

+
+ {% endif %} + + +
+
+
+
+
+
+ Security Features +
+
    +
  • Automatic location tracking
    • +
  • Suspicious login detection
    • +
  • Email alerts for new locations
    • +
  • Two-factor authentication
    • +
+
+
+
+
+
+
+
+ Security Tips +
+
    +
  • Review this page regularly
    • +
  • Report suspicious activity
    • +
  • Use strong, unique passwords
    • +
  • Keep your devices secure
    • +
+
+
+
+
+
+ + +
+ + Back to Dashboard + + + Security Settings + +
+
+
+
+
+{% endblock %} + +{% block scripts %} + +{% endblock %} diff --git a/app/templates/base.html b/app/templates/base.html index 893b04d..c66b4bf 100644 --- a/app/templates/base.html +++ b/app/templates/base.html @@ -75,9 +75,13 @@
  • {{ current_user.username }} - -
      -
    • Profile
      • + diff --git a/app/templates/emails/location_approved.html b/app/templates/emails/location_approved.html new file mode 100644 index 0000000..146de31 --- /dev/null +++ b/app/templates/emails/location_approved.html @@ -0,0 +1,82 @@ + + + + + + Login Location Approved + + + +
      +
      +
      ✅
      +

      Location Approved Successfully

      +
      + +

      Hello {{ user_name }},

      + +

      Thank you for approving the new login location for your {{ app_name }} account.

      + +
      +

      📍 Approved Location:

      +
        +
      • Location: {{ location.city or 'Unknown' }}{% if location.region %}, {{ location.region }}{% endif %}{% if location.country %}, {{ location.country }}{% endif %}
        • +
      • Approved at: {{ approval_time }}
        • +
      +
      + +

      This location has been added to your list of trusted locations. Future logins from this location will not require additional approval.

      + +

      Security Reminder: If you did not approve this location, please contact support immediately and change your password.

      + +
      +

      This email was sent from {{ app_name }} security system.

      +

      © 2025 {{ app_name }}. All rights reserved.

      +
      +
      + + diff --git a/app/templates/emails/suspicious_login.html b/app/templates/emails/suspicious_login.html new file mode 100644 index 0000000..62ae9ac --- /dev/null +++ b/app/templates/emails/suspicious_login.html @@ -0,0 +1,129 @@ + + + + + + + Suspicious Login Detected + + + +
      +
      +
      🔐
      +

      Suspicious Login Detected

      +
      + +

      Hello {{ user_name }},

      + +

      We detected a login to your {{ app_name }} account from a new or unusual location. If this was you, please approve this location by clicking the button below.

      + +
      +

      📍 Login Details:

      +
        +
      • Location: {{ location.city or 'Unknown' }}{% if location.region %}, {{ location.region }}{% endif %}{% if location.country %}, {{ location.country }}{% endif %}
        • +
      • Time: {{ login_time }}
        • +
      • IP Address: {{ location.ip_address or 'Hidden for privacy' }}
        • + {% if location.distance_km %} +
      • Distance from last login: {{ location.distance_km }} km
        • + {% endif %} +
      +
      + +

      ⚠️ Action Required

      +

      If this login was authorized by you, please approve this location by clicking the button below:

      + + + +
      +

      🛡️ Security Notice:

      +
        +
      • If you did not attempt to log in, your account may be compromised
        • +
      • Change your password immediately if this login was not authorized
        • +
      • Enable two-factor authentication for additional security
        • +
      • This approval link expires in 24 hours
        • +
      +
      + +

      If you did not make this login attempt:

      +
        +
      1. Do not click the approval button
        2. +
      2. Change your password immediately
        3. +
      3. Review your account for any suspicious activity
        4. +
      4. Contact support if you need assistance
        5. +
      + +
      +

      This email was sent from {{ app_name }} security system.

      +

      For your security, do not forward this email or share the approval link.

      +

      © 2025 {{ app_name }}. All rights reserved.

      +
      +
      + + diff --git a/app/utils/__init__.py b/app/utils/__init__.py new file mode 100644 index 0000000..8b144e7 --- /dev/null +++ b/app/utils/__init__.py @@ -0,0 +1,13 @@ +""" +Utility modules for Flask 2FA application. + +This package contains reusable utility functions for: +- Email sending and notification services +- Location tracking and geolocation services +- Security utilities and token management +""" + +from .mail import MailService +from .location import LocationService + +__all__ = ['MailService', 'LocationService'] diff --git a/app/utils/location.py b/app/utils/location.py new file mode 100644 index 0000000..601abec --- /dev/null +++ b/app/utils/location.py @@ -0,0 +1,338 @@ +""" +Location tracking and geolocation service for Flask 2FA application. + +This module provides location tracking functionality including: +- IP geolocation lookup +- Distance calculation between locations +- Suspicious login detection +- Location approval management + +Security features: +- Rate limiting for API calls +- Error handling and fallbacks +- Privacy-aware location tracking +""" + +import requests +import logging +import math +from typing import Optional, Dict, Any, Tuple +from flask import request, current_app +from datetime import datetime, timedelta + + +class LocationService: + """ + Service for handling location tracking and geolocation functionality. + + Provides methods for IP geolocation, distance calculation, + and suspicious login detection. + """ + + def __init__(self): + """Initialize location service.""" + self.logger = logging.getLogger(__name__) + self.api_timeout = 5 # seconds + self.cache = {} # Simple in-memory cache + self.cache_duration = timedelta(hours=1) + + def get_client_ip(self) -> str: + """ + Get client IP address from request. + + Handles various proxy headers for accurate IP detection. + + Returns: + str: Client IP address + """ + # Check for forwarded headers (proxy/load balancer) + forwarded_for = request.headers.get('X-Forwarded-For') + if forwarded_for: + # Take first IP if multiple are present + return forwarded_for.split(',')[0].strip() + + real_ip = request.headers.get('X-Real-IP') + if real_ip: + return real_ip + + # Fallback to direct connection + return request.remote_addr or '127.0.0.1' + + def get_user_agent(self) -> str: + """ + Get user agent string from request. + + Returns: + str: User agent string + """ + return request.headers.get('User-Agent', 'Unknown') + + def get_location_from_ip(self, ip_address: str) -> Dict[str, Any]: + """ + Get location data from IP address using geolocation API. + + Uses multiple fallback services for reliability. + + Args: + ip_address: IP address to lookup + + Returns: + Dict containing location data + """ + # Skip private/local IPs + if self._is_private_ip(ip_address): + return self._get_default_location("Private Network") + + # Check cache first + cache_key = f"location_{ip_address}" + if cache_key in self.cache: + cached_data, timestamp = self.cache[cache_key] + if datetime.utcnow() - timestamp < self.cache_duration: + return cached_data + + location_data = self._fetch_location_data(ip_address) + + # Cache the result + self.cache[cache_key] = (location_data, datetime.utcnow()) + + return location_data + + def _fetch_location_data(self, ip_address: str) -> Dict[str, Any]: + """ + Fetch location data from external APIs with fallbacks. + + Args: + ip_address: IP address to lookup + + Returns: + Dict containing location data + """ + # Try multiple geolocation services + services = [ + self._fetch_from_ipapi, + self._fetch_from_ipinfo, + self._fetch_from_freeipapi + ] + + for service in services: + try: + data = service(ip_address) + if data and data.get('country'): + self.logger.info(f"Successfully fetched location for {ip_address}") + return data + except Exception as e: + self.logger.warning(f"Failed to fetch from service: {str(e)}") + continue + + # All services failed + self.logger.error(f"All geolocation services failed for {ip_address}") + return self._get_default_location("Unknown") + + def _fetch_from_ipapi(self, ip_address: str) -> Dict[str, Any]: + """Fetch location from ip-api.com (free service).""" + url = f"http://ip-api.com/json/{ip_address}" + response = requests.get(url, timeout=self.api_timeout) + response.raise_for_status() + + data = response.json() + if data.get('status') == 'success': + return { + 'country': data.get('country'), + 'region': data.get('regionName'), + 'city': data.get('city'), + 'latitude': data.get('lat'), + 'longitude': data.get('lon'), + 'timezone': data.get('timezone'), + 'isp': data.get('isp') + } + raise Exception(f"API returned error: {data.get('message')}") + + def _fetch_from_ipinfo(self, ip_address: str) -> Dict[str, Any]: + """Fetch location from ipinfo.io (free tier).""" + url = f"https://ipinfo.io/{ip_address}/json" + response = requests.get(url, timeout=self.api_timeout) + response.raise_for_status() + + data = response.json() + if 'country' in data: + loc = data.get('loc', '').split(',') + return { + 'country': data.get('country'), + 'region': data.get('region'), + 'city': data.get('city'), + 'latitude': float(loc[0]) if len(loc) > 0 and loc[0] else None, + 'longitude': float(loc[1]) if len(loc) > 1 and loc[1] else None, + 'timezone': data.get('timezone'), + 'isp': data.get('org') + } + raise Exception("No location data in response") + + def _fetch_from_freeipapi(self, ip_address: str) -> Dict[str, Any]: + """Fetch location from freeipapi.com.""" + url = f"https://freeipapi.com/api/json/{ip_address}" + response = requests.get(url, timeout=self.api_timeout) + response.raise_for_status() + + data = response.json() + if data.get('countryName'): + return { + 'country': data.get('countryName'), + 'region': data.get('regionName'), + 'city': data.get('cityName'), + 'latitude': data.get('latitude'), + 'longitude': data.get('longitude'), + 'timezone': data.get('timeZone'), + 'isp': None + } + raise Exception("No location data in response") + + def _is_private_ip(self, ip_address: str) -> bool: + """ + Check if IP address is private/local. + + Args: + ip_address: IP address to check + + Returns: + bool: True if IP is private + """ + private_ranges = [ + '127.', # Localhost + '10.', # Private network + '172.16.', # Private network + '172.17.', # Private network + '172.18.', # Private network + '172.19.', # Private network + '172.20.', # Private network + '172.21.', # Private network + '172.22.', # Private network + '172.23.', # Private network + '172.24.', # Private network + '172.25.', # Private network + '172.26.', # Private network + '172.27.', # Private network + '172.28.', # Private network + '172.29.', # Private network + '172.30.', # Private network + '172.31.', # Private network + '192.168.', # Private network + '::1', # IPv6 localhost + ] + + return any(ip_address.startswith(prefix) for prefix in private_ranges) + + def _get_default_location(self, reason: str) -> Dict[str, Any]: + """ + Get default location data when lookup fails. + + Args: + reason: Reason for using default location + + Returns: + Dict containing default location data + """ + return { + 'country': reason, + 'region': None, + 'city': None, + 'latitude': None, + 'longitude': None, + 'timezone': None, + 'isp': None + } + + def calculate_distance(self, + lat1: float, lon1: float, + lat2: float, lon2: float) -> float: + """ + Calculate distance between two geographic points using Haversine formula. + + Args: + lat1, lon1: First point coordinates + lat2, lon2: Second point coordinates + + Returns: + float: Distance in kilometers + """ + if None in [lat1, lon1, lat2, lon2]: + return 0.0 + + # Convert to radians + lat1, lon1, lat2, lon2 = map(math.radians, [lat1, lon1, lat2, lon2]) + + # Haversine formula + dlat = lat2 - lat1 + dlon = lon2 - lon1 + a = (math.sin(dlat/2)**2 + + math.cos(lat1) * math.cos(lat2) * math.sin(dlon/2)**2) + c = 2 * math.asin(math.sqrt(a)) + + # Radius of Earth in kilometers + r = 6371 + + return c * r + + def is_suspicious_location(self, + user_id: int, + new_location: Dict[str, Any]) -> Tuple[bool, Optional[Dict[str, Any]]]: + """ + Check if login location is suspicious based on user's history. + + Args: + user_id: User ID + new_location: New location data + + Returns: + Tuple of (is_suspicious: bool, last_location: Dict or None) + """ + from app.models import LoginLocation + + # Get user's last approved login location + last_login = LoginLocation.query.filter_by( + user_id=user_id, + is_approved=True + ).order_by(LoginLocation.login_time.desc()).first() + + if not last_login: + # First login is always approved + return False, None + + # Skip distance check if coordinates are missing + if (not all([new_location.get('latitude'), new_location.get('longitude')]) or + not all([last_login.latitude, last_login.longitude])): + return False, { + 'country': last_login.country, + 'region': last_login.region, + 'city': last_login.city, + 'login_time': last_login.login_time + } + + # Calculate distance + distance = self.calculate_distance( + last_login.latitude, last_login.longitude, + new_location['latitude'], new_location['longitude'] + ) + + # Check if distance exceeds threshold + threshold = current_app.config.get('SUSPICIOUS_LOGIN_THRESHOLD_KM', 100) + is_suspicious = distance > threshold + + last_location_data = { + 'country': last_login.country, + 'region': last_login.region, + 'city': last_login.city, + 'login_time': last_login.login_time, + 'distance_km': round(distance, 2) + } + + if is_suspicious: + self.logger.warning( + f"Suspicious login detected for user {user_id}: " + f"{distance:.2f}km from last location" + ) + + return is_suspicious, last_location_data + + +# Global location service instance +location_service = LocationService() diff --git a/app/utils/mail.py b/app/utils/mail.py new file mode 100644 index 0000000..f6bc3a1 --- /dev/null +++ b/app/utils/mail.py @@ -0,0 +1,204 @@ +""" +Modular mail service for Flask 2FA application. + +This module provides a reusable mail service that can be used for: +- Suspicious login alerts +- Location approval requests +- Account security notifications +- General application notifications + +Security features: +- HTML email with security headers +- Secure token handling +- Rate limiting support +- Template-based emails +""" + +import logging +from flask import current_app, render_template, url_for +from flask_mail import Mail, Message +from datetime import datetime, timedelta +from typing import Optional, List, Dict, Any + + +class MailService: + """ + Centralized mail service for the application. + + Provides secure email functionality with template support + and security best practices. + """ + + def __init__(self, mail_instance: Optional[Mail] = None): + """ + Initialize mail service. + + Args: + mail_instance: Flask-Mail instance (optional) + """ + self.mail = mail_instance + self.logger = logging.getLogger(__name__) + + def init_app(self, app, mail_instance: Mail): + """Initialize mail service with Flask app and Mail instance.""" + self.mail = mail_instance + with app.app_context(): + self.logger.info("Mail service initialized") + + def send_email(self, + subject: str, + recipients: List[str], + template: str, + **template_vars) -> bool: + """ + Send an email using template. + + Args: + subject: Email subject line + recipients: List of recipient email addresses + template: Template name (without .html extension) + **template_vars: Variables to pass to template + + Returns: + bool: True if email sent successfully, False otherwise + """ + try: + # Create message + msg = Message( + subject=subject, + recipients=recipients, + sender=current_app.config['MAIL_DEFAULT_SENDER'] + ) + + # Render HTML template + msg.html = render_template(f'emails/{template}.html', **template_vars) + + # Add security headers to email + msg.extra_headers = { + 'X-Mailer': 'Flask-2FA-App', + 'X-Priority': '1', + 'X-MSMail-Priority': 'High' + } + + # Send email + self.mail.send(msg) + + self.logger.info(f"Email sent successfully to {recipients}: {subject}") + return True + + except Exception as e: + self.logger.error(f"Failed to send email to {recipients}: {str(e)}") + return False + + def send_suspicious_login_alert(self, + user_email: str, + user_name: str, + location_data: Dict[str, Any], + approval_token: str) -> bool: + """ + Send suspicious login alert email. + + Args: + user_email: User's email address + user_name: User's display name + location_data: Dictionary containing location information + approval_token: Token for approving the location + + Returns: + bool: True if email sent successfully + """ + approval_url = url_for('auth.approve_location', + token=approval_token, + _external=True) + + return self.send_email( + subject="🔐 Suspicious Login Detected - Action Required", + recipients=[user_email], + template="suspicious_login", + user_name=user_name, + location=location_data, + approval_url=approval_url, + login_time=datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S UTC'), + app_name="Flask 2FA App" + ) + + def send_location_approved_notification(self, + user_email: str, + user_name: str, + location_data: Dict[str, Any]) -> bool: + """ + Send location approval confirmation email. + + Args: + user_email: User's email address + user_name: User's display name + location_data: Dictionary containing location information + + Returns: + bool: True if email sent successfully + """ + return self.send_email( + subject="✅ Login Location Approved", + recipients=[user_email], + template="location_approved", + user_name=user_name, + location=location_data, + approval_time=datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S UTC'), + app_name="Flask 2FA App" + ) + + def send_account_security_alert(self, + user_email: str, + user_name: str, + alert_type: str, + details: Dict[str, Any]) -> bool: + """ + Send general account security alert. + + Args: + user_email: User's email address + user_name: User's display name + alert_type: Type of security alert + details: Additional details for the alert + + Returns: + bool: True if email sent successfully + """ + return self.send_email( + subject=f"🔔 Security Alert: {alert_type}", + recipients=[user_email], + template="security_alert", + user_name=user_name, + alert_type=alert_type, + details=details, + timestamp=datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S UTC'), + app_name="Flask 2FA App" + ) + + def send_welcome_email(self, + user_email: str, + user_name: str) -> bool: + """ + Send welcome email to new users. + + Args: + user_email: User's email address + user_name: User's display name + + Returns: + bool: True if email sent successfully + """ + dashboard_url = url_for('main.dashboard', _external=True) + + return self.send_email( + subject="🎉 Welcome to Flask 2FA App!", + recipients=[user_email], + template="welcome", + user_name=user_name, + dashboard_url=dashboard_url, + app_name="Flask 2FA App" + ) + + +# Global mail service instance +mail_service = MailService() diff --git a/config.py b/config.py index db5283d..4c003f9 100644 --- a/config.py +++ b/config.py @@ -24,9 +24,21 @@ class Config: SESSION_COOKIE_SECURE = True # HTTPS only cookies SESSION_COOKIE_HTTPONLY = True # Prevent XSS attacks SESSION_COOKIE_SAMESITE = 'Lax' # CSRF protection - - # Additional security headers + # Additional security headers SEND_FILE_MAX_AGE_DEFAULT = timedelta(hours=1) + + # Mail configuration + MAIL_SERVER = os.environ.get('MAIL_SERVER') or 'smtp.gmail.com' + MAIL_PORT = int(os.environ.get('MAIL_PORT') or 587) + MAIL_USE_TLS = os.environ.get('MAIL_USE_TLS', 'true').lower() in ['true', 'on', '1'] + MAIL_USERNAME = os.environ.get('MAIL_USERNAME') + MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD') + MAIL_DEFAULT_SENDER = os.environ.get('MAIL_DEFAULT_SENDER') or 'noreply@flask2fa.com' + + # Security settings for location tracking + MAX_LOGIN_ATTEMPTS = 5 + LOCATION_APPROVAL_TOKEN_EXPIRE = timedelta(hours=24) + SUSPICIOUS_LOGIN_THRESHOLD_KM = 100 # Alert if login from >100km away class DevelopmentConfig(Config): diff --git a/requirements.txt b/requirements.txt index 0e08652..775c738 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,9 +3,12 @@ Flask-Login>=0.6.3 Flask-WTF>=1.2.1 Flask-SQLAlchemy>=3.1.1 Flask-Migrate>=4.0.5 +Flask-Mail>=0.9.1 SQLAlchemy>=2.0.0 PyOTP>=2.9.0 bcrypt>=4.1.2 qrcode[pil]>=7.4.2 python-dotenv>=1.0.0 WTForms>=3.1.0 +requests>=2.31.0 +email-validator>=2.1.0