"""Security dashboard service. The single home for the Security page's read/write logic, built entirely on the existing infrastructure: * **Persistence** reuses the one audit sink — :func:`app.auth.services.log_auth_event` dual-writes a :class:`~app.models.auth.SecurityEvent` row here, so login history, recent activity, failed-login monitoring and suspicious-activity all read from events the app already emits (no new call sites). * **Device/IP metadata** reuses :mod:`app.auth.device_info` (UA parser + IP mask) and the salted-hash / client-IP helpers already used for view tracking. * **Email** reuses :func:`app.auth.otp_service.send_email` and the branded layout. Privacy: raw IPs are never stored — only a masked display string and a salted hash. Location is best-effort ("Location unavailable" until a GeoIP source is configured), per the approved graceful-degrade approach. """ from __future__ import annotations import json from datetime import timedelta from flask import current_app, render_template, url_for from app.auth import otp_service from app.auth.device_info import mask_ip, parse_user_agent from app.auth.services import AuthEvent, LOGIN_EVENTS, result_for from app.extensions import db from app.marketplace.services import _client_ip, _hash_value from app.models import SecurityEvent, TrustedDevice from app.models.base import utcnow LOCATION_UNAVAILABLE = "Location unavailable" # event key -> (human label, bootstrap icon, tone). Drives the activity feed and # timeline; one place to add a new event's presentation. EVENT_CATALOG: dict[str, tuple[str, str, str]] = { AuthEvent.LOGIN_SUCCESS: ("Successful sign-in", "bi-box-arrow-in-right", "success"), AuthEvent.LOGIN_OTP_SUCCESS: ("Signed in with email code", "bi-shield-check", "success"), AuthEvent.LOGIN_FAILED: ("Failed sign-in attempt", "bi-x-octagon", "danger"), AuthEvent.LOGIN_OTP_FAILURE: ("Incorrect verification code", "bi-x-octagon", "danger"), AuthEvent.ACCOUNT_LOCKED: ("Account temporarily locked", "bi-lock-fill", "danger"), AuthEvent.LOGIN_OTP_SENT: ("Verification code sent", "bi-envelope", "info"), AuthEvent.LOGIN_OTP_RESEND: ("Verification code resent", "bi-envelope", "info"), AuthEvent.TRUSTED_DEVICE_CREATED: ("New trusted device added", "bi-laptop", "primary"), AuthEvent.TRUSTED_DEVICE_USED: ("Signed in from a trusted device", "bi-laptop", "info"), AuthEvent.TRUSTED_DEVICE_REVOKED: ("Trusted device removed", "bi-trash", "warning"), AuthEvent.TRUSTED_DEVICE_RENAMED: ("Trusted device renamed", "bi-pencil", "info"), AuthEvent.TRUSTED_DEVICES_REVOKED_AFTER_PASSWORD_RESET: ("Trusted devices cleared after password reset", "bi-shield-exclamation", "warning"), AuthEvent.ACCOUNT_CREATED: ("Account created", "bi-person-plus-fill", "info"), AuthEvent.EMAIL_VERIFIED: ("Email verified", "bi-patch-check-fill", "success"), AuthEvent.PASSWORD_CHANGED: ("Password changed", "bi-key-fill", "primary"), AuthEvent.PASSWORD_RESET_COMPLETED: ("Password reset completed", "bi-key-fill", "warning"), AuthEvent.TWO_FACTOR_ENABLED: ("Two-factor authentication enabled", "bi-shield-lock-fill", "success"), AuthEvent.TWO_FACTOR_DISABLED: ("Two-factor authentication disabled", "bi-shield-slash", "warning"), AuthEvent.SUSPICIOUS_LOGIN: ("Unrecognized sign-in", "bi-exclamation-triangle-fill", "danger"), } def describe_event(event: str) -> dict: label, icon, tone = EVENT_CATALOG.get(event, (event.replace("_", " ").title(), "bi-shield", "secondary")) return {"label": label, "icon": icon, "tone": tone} # --- Human-readable method labels ---------------------------------------------- _METHOD_LABELS = { "password": "Password only", "email_otp": "Email OTP", "trusted_device": "Trusted device", } def method_label(method: str | None) -> str: return _METHOD_LABELS.get(method or "", "—") # --- Write path (called from log_auth_event; best-effort) ---------------------- def record_security_event(user, event, *, request=None, ip=None, method=None, result=None, detail=None, location=None) -> SecurityEvent | None: """Persist one security event for ``user``. Best-effort: never raises into the caller (an audit failure must not break login). Parses coarse device fields from the request UA and stores only a masked IP + salted hash.""" ua = request.headers.get("User-Agent", "") if request is not None else "" info = parse_user_agent(ua) client_ip = ip or (_client_ip(request) if request is not None else None) row = SecurityEvent( user_id=user.id, event=event, result=result or result_for(event), method=method, browser=info["browser"] if ua else None, os=info["os"] if ua else None, device_type=info["device_type"] if ua else None, ip_display=mask_ip(client_ip), ip_hash=_hash_value(client_ip), location=location or (LOCATION_UNAVAILABLE if client_ip else None), detail=(str(detail)[:255] if detail else None), ) try: db.session.add(row) db.session.commit() return row except Exception: # pragma: no cover - defensive db.session.rollback() current_app.logger.exception("record_security_event failed for %s", event) return None # --- Read path ----------------------------------------------------------------- def get_login_history(user, page: int = 1, per_page: int = 20): """Paginated sign-in events (newest first) for the Login History view.""" query = ( db.select(SecurityEvent) .where(SecurityEvent.user_id == user.id, SecurityEvent.event.in_(list(LOGIN_EVENTS))) .order_by(SecurityEvent.created_at.desc(), SecurityEvent.id.desc()) ) return db.paginate(query, page=page, per_page=per_page, error_out=False) def get_recent_activity(user, limit: int = 12) -> list[dict]: """Most recent security events (any type) as presentation dicts.""" rows = db.session.scalars( db.select(SecurityEvent) .where(SecurityEvent.user_id == user.id) .order_by(SecurityEvent.created_at.desc(), SecurityEvent.id.desc()) .limit(limit) ).all() items = [] for r in rows: meta = describe_event(r.event) items.append({ "icon": meta["icon"], "tone": meta["tone"], "label": meta["label"], "timestamp": r.created_at, "ip_display": r.ip_display, "browser": r.browser, "os": r.os, "location": r.location, }) return items def get_failed_login_summary(user, hours: int = 24) -> dict: """Count + last-attempt time of failed sign-ins in the trailing window.""" since = utcnow() - timedelta(hours=hours) rows = db.session.scalars( db.select(SecurityEvent) .where( SecurityEvent.user_id == user.id, SecurityEvent.event == AuthEvent.LOGIN_FAILED, SecurityEvent.created_at >= since, ) .order_by(SecurityEvent.created_at.desc()) ).all() return {"count": len(rows), "last_at": rows[0].created_at if rows else None, "hours": hours} def _latest_event_at(user, event: str): row = db.session.scalar( db.select(SecurityEvent) .where(SecurityEvent.user_id == user.id, SecurityEvent.event == event) .order_by(SecurityEvent.created_at.desc()) .limit(1) ) return row.created_at if row else None # --- Expanded authentication timeline ------------------------------------------ def build_auth_timeline(user) -> list[dict]: """Account-security milestones (newest-relevant first is handled in the template). Combines the user's own columns with persisted events so newer facts (2FA, trusted devices) appear without duplicating state.""" timeline: list[dict] = [] def add(icon, tone, label, timestamp, note=None): timeline.append({"icon": icon, "tone": tone, "label": label, "timestamp": timestamp, "note": note}) add("bi-person-plus-fill", "info", "Account created", user.created_at) if user.email_verified: add("bi-patch-check-fill", "success", "Email verified", user.verified_at, None if user.verified_at else "Verified") else: add("bi-exclamation-triangle-fill", "warning", "Email not verified", None, "Pending") add("bi-key-fill", "secondary", "Password changed", user.last_password_change, None if user.last_password_change else "Never changed") reset_at = _latest_event_at(user, AuthEvent.PASSWORD_RESET_COMPLETED) if reset_at: add("bi-key-fill", "warning", "Password reset", reset_at) if user.two_factor_enabled: add("bi-shield-lock-fill", "success", "Two-factor enabled", _latest_event_at(user, AuthEvent.TWO_FACTOR_ENABLED)) else: disabled_at = _latest_event_at(user, AuthEvent.TWO_FACTOR_DISABLED) if disabled_at: add("bi-shield-slash", "warning", "Two-factor disabled", disabled_at) device_at = _latest_event_at(user, AuthEvent.TRUSTED_DEVICE_CREATED) if device_at: add("bi-laptop", "primary", "Trusted device added", device_at) removed_at = _latest_event_at(user, AuthEvent.TRUSTED_DEVICE_REVOKED) if removed_at: add("bi-trash", "secondary", "Trusted device removed", removed_at) new_device_login = _latest_event_at(user, AuthEvent.SUSPICIOUS_LOGIN) if new_device_login: add("bi-exclamation-triangle-fill", "danger", "Login from a new device", new_device_login) add("bi-box-arrow-in-right", "primary", "Last login", user.last_login, None if user.last_login else "No sign-in recorded yet") # Sort newest-first, keeping entries without a timestamp at the end. timeline.sort(key=lambda e: (e["timestamp"] is not None, e["timestamp"] or utcnow()), reverse=True) return timeline # --- Security score ------------------------------------------------------------- def compute_security_score(user) -> dict: """A 0–100 score from signals the app already tracks, with per-factor status and improvement suggestions. Weights sum to 100.""" now = utcnow() active_devices = db.session.scalar( db.select(db.func.count(TrustedDevice.id)).where( TrustedDevice.user_id == user.id, TrustedDevice.revoked_at.is_(None), TrustedDevice.expires_at > now, ) ) or 0 recent_password = bool( user.last_password_change and user.last_password_change > now - timedelta(days=180) ) suspicious = db.session.scalar( db.select(db.func.count(SecurityEvent.id)).where( SecurityEvent.user_id == user.id, SecurityEvent.event.in_([AuthEvent.SUSPICIOUS_LOGIN, AuthEvent.LOGIN_FAILED]), SecurityEvent.created_at >= now - timedelta(days=30), ) ) or 0 factors = [ {"key": "email", "label": "Email verified", "weight": 25, "ok": bool(user.email_verified), "suggestion": "Verify your email address."}, {"key": "twofa", "label": "Two-factor authentication enabled", "weight": 25, "ok": bool(user.two_factor_enabled), "suggestion": "Turn on email two-factor authentication."}, {"key": "password", "label": "Password changed in the last 6 months", "weight": 20, "ok": recent_password, "suggestion": "Update your password to keep it fresh."}, {"key": "activity", "label": "No recent suspicious or failed sign-ins", "weight": 15, "ok": (suspicious == 0), "suggestion": "Review your recent sign-in activity."}, {"key": "devices", "label": "Trusted devices reviewed", "weight": 15, "ok": (active_devices <= 5), "suggestion": "Remove trusted devices you no longer use."}, ] score = sum(f["weight"] for f in factors if f["ok"]) suggestions = [f["suggestion"] for f in factors if not f["ok"]] if score >= 85: tone, label = "success", "Strong" elif score >= 60: tone, label = "warning", "Good" else: tone, label = "danger", "Needs attention" return {"score": score, "tone": tone, "label": label, "factors": factors, "suggestions": suggestions} # --- Suspicious-activity detection --------------------------------------------- def detect_new_context(user, request) -> list[str]: """Compare this sign-in's device/IP against the account's prior **successful** sign-ins. Returns flags like ``new_browser`` / ``new_device`` / ``new_ip``. Empty for a first-ever sign-in (nothing to compare against) so a new account is never flagged as suspicious.""" prior = db.session.scalars( db.select(SecurityEvent).where( SecurityEvent.user_id == user.id, SecurityEvent.event.in_([AuthEvent.LOGIN_SUCCESS, AuthEvent.LOGIN_OTP_SUCCESS]), ) ).all() if not prior: return [] ua = request.headers.get("User-Agent", "") if request is not None else "" info = parse_user_agent(ua) ip_hash = _hash_value(_client_ip(request)) if request is not None else None seen_browsers = {(p.browser, p.os) for p in prior} seen_devices = {p.device_type for p in prior} seen_ips = {p.ip_hash for p in prior if p.ip_hash} flags = [] if ua and (info["browser"], info["os"]) not in seen_browsers: flags.append("new_browser") if ua and info["device_type"] not in seen_devices: flags.append("new_device") if ip_hash and ip_hash not in seen_ips: flags.append("new_ip") return flags _FLAG_TEXT = { "new_browser": "a new browser", "new_device": "a new device", "new_ip": "a new network / IP address", } def flags_to_text(flags: list[str]) -> str: parts = [_FLAG_TEXT[f] for f in flags if f in _FLAG_TEXT] if not parts: return "an unrecognized context" if len(parts) == 1: return parts[0] return ", ".join(parts[:-1]) + " and " + parts[-1] # --- Active sessions (approximate) --------------------------------------------- def get_active_sessions(user, request) -> list[dict]: """Approximate 'active sessions' from trusted devices plus this browser. Flask-Login is stateless, so this is an honest approximation: each trusted device is shown as a recognised session, and the current browser is always listed (even if not trusted). 'Sign out other devices' revokes the other trusted devices (they'll need an OTP again).""" from app.auth.trusted_devices import current_device_id current_id = current_device_id(user, request) now = utcnow() devices = db.session.scalars( db.select(TrustedDevice) .where( TrustedDevice.user_id == user.id, TrustedDevice.revoked_at.is_(None), TrustedDevice.expires_at > now, ) .order_by(TrustedDevice.last_used_at.desc()) ).all() sessions = [] for d in devices: sessions.append({ "id": d.id, "name": d.display_name, "last_used": d.last_used_at, "is_current": d.id == current_id, "trusted": True, }) # If the current browser isn't a trusted device, surface it as an untrusted # current session so the user always sees "this device". if current_id is None: info = parse_user_agent(request.headers.get("User-Agent", "") if request else "") sessions.insert(0, { "id": None, "name": info["label"], "last_used": user.last_login, "is_current": True, "trusted": False, }) return sessions # --- Email notification preferences -------------------------------------------- DEFAULT_NOTIFICATION_PREFS = { "new_login": True, "new_device": True, "password_changed": True, "two_factor": True, "failed_login": True, "alerts": True, } NOTIFICATION_LABELS = { "new_login": "New sign-in to your account", "new_device": "A new device is trusted", "password_changed": "Your password is changed", "two_factor": "Two-factor is enabled or disabled", "failed_login": "Repeated failed sign-in attempts", "alerts": "Security alerts (suspicious activity)", } def get_notification_prefs(user) -> dict: prefs = dict(DEFAULT_NOTIFICATION_PREFS) raw = getattr(user, "security_notification_prefs", None) if raw: try: stored = json.loads(raw) if isinstance(stored, dict): for k in prefs: if k in stored: prefs[k] = bool(stored[k]) except (ValueError, TypeError): pass return prefs def set_notification_prefs(user, prefs: dict) -> None: clean = {k: bool(prefs.get(k, DEFAULT_NOTIFICATION_PREFS[k])) for k in DEFAULT_NOTIFICATION_PREFS} user.security_notification_prefs = json.dumps(clean) db.session.commit() def notifications_enabled(user, key: str) -> bool: return get_notification_prefs(user).get(key, True) # --- Security email notifications (reuse otp_service.send_email) ---------------- def send_security_email(user, subject: str, heading: str, intro: str, lines: list[str] | None = None, cta_label=None, cta_url=None) -> bool: """Render + send a branded security notification. Best-effort; reuses the shared sender and the security_alert email templates.""" try: ctx = { "heading": heading, "intro": intro, "lines": lines or [], "cta_label": cta_label, "cta_url": cta_url, "support_email": current_app.config.get("SUPPORT_EMAIL", "support@studentbazaar.in"), } return otp_service.send_email( user.email, subject, render_template("emails/security_alert.txt", **ctx), render_template("emails/security_alert.html", **ctx), label="security_alert", ) except Exception: # pragma: no cover current_app.logger.exception("Failed to send security email to user %s", getattr(user, "id", None)) return False def _security_url() -> str | None: try: return url_for("auth.security", _external=True) except Exception: return None def notify_new_login(user, request, flags: list[str]) -> None: """Email the user about an **unrecognized** sign-in (new browser/device/IP). Routine and trusted sign-ins are intentionally silent (no per-login spam). Governed by the ``alerts`` or ``new_login`` preference.""" prefs = get_notification_prefs(user) if not (prefs.get("alerts") or prefs.get("new_login")): return info = parse_user_agent(request.headers.get("User-Agent", "") if request else "") when = utcnow().strftime("%d %b %Y, %H:%M UTC") where = f"{info['browser']} on {info['os']} ({info['device_type']})" send_security_email( user, "Unrecognized sign-in to your CampusCart account", "Was this you?", f"We noticed a sign-in from {flags_to_text(flags)}.", [f"Device: {where}", f"Time: {when}", "If this was you, no action is needed.", "If not, reset your password and remove trusted devices immediately."], "Review security", _security_url(), ) def notify_password_changed(user) -> None: if not notifications_enabled(user, "password_changed"): return send_security_email( user, "Your CampusCart password was changed", "Password changed", "Your account password was just changed.", ["If you made this change, no action is needed.", "If you didn't, reset your password immediately and contact support."], "Review security", _security_url(), ) def notify_two_factor(user, enabled: bool) -> None: if not notifications_enabled(user, "two_factor"): return state = "enabled" if enabled else "disabled" send_security_email( user, f"Two-factor authentication {state}", f"Two-factor {state}", f"Email two-factor authentication was {state} on your account.", ["If this wasn't you, secure your account right away."], "Review security", _security_url(), )