admin / Synapse-NetscanXi

"""
Authentication & authorization (v4).

User accounts with hashed passwords and four roles (v9.1):
    admin        - full control across ALL tenants, incl. global user mgmt
    tenant_admin - full control WITHIN their own tenant (users, schedules,
                   scans, settings, integrations) but never across tenants
    operator     - can run/cancel scans + everything a viewer can
    viewer       - read-only (dashboard, assets, history, export)

Password hashing uses werkzeug.security (ships with Flask -> no new deps).

Backwards compatibility: the legacy single shared password (NETSCAN_PASSWORD)
still works as a fallback "admin" login when no user accounts exist, so existing
deployments keep working until an admin account is created.

First-admin bootstrap: on startup, if there are no users and the env vars
NETSCAN_ADMIN_USER / NETSCAN_ADMIN_PASSWORD are set, an admin account is created
automatically so you are never locked out of a fresh install.
"""

import os
import secrets

import pyotp
from werkzeug.security import generate_password_hash, check_password_hash

from . import db

MFA_ISSUER = os.environ.get("NETSCAN_MFA_ISSUER", "NetScan Xi")
BACKUP_CODE_COUNT = 10

# Role hierarchy: higher number = more privilege.
# v9.1: tenant_admin sits between operator and (global) admin.
ROLE_LEVEL = {"viewer": 1, "operator": 2, "tenant_admin": 3, "admin": 4}
DEFAULT_ROLE = "viewer"

LEGACY_PASSWORD = os.environ.get("NETSCAN_PASSWORD", "")


def hash_password(password):
    return generate_password_hash(password)


def verify_password(password_hash, password):
    try:
        return check_password_hash(password_hash, password)
    except Exception:
        return False


def role_at_least(role, required):
    """True if `role` meets or exceeds the `required` role level."""
    return ROLE_LEVEL.get(role, 0) >= ROLE_LEVEL.get(required, 99)


def accounts_exist(db_path=db.DEFAULT_DB_PATH):
    return db.count_users(db_path=db_path) > 0


def bootstrap_admin(db_path=db.DEFAULT_DB_PATH):
    """
    Ensure there is a way in on a fresh install.
    If no users exist and NETSCAN_ADMIN_USER/PASSWORD are provided, create an
    admin. Returns the created username or None.
    """
    if accounts_exist(db_path=db_path):
        return None
    user = os.environ.get("NETSCAN_ADMIN_USER", "").strip()
    pw = os.environ.get("NETSCAN_ADMIN_PASSWORD", "")
    if user and pw:
        uid = db.create_user(user, hash_password(pw), role="admin", db_path=db_path)
        if uid:
            return user
    return None


# ---------------------------------------------------------------------------
# MFA (TOTP authenticator app + one-time backup codes)
# ---------------------------------------------------------------------------
def new_totp_secret():
    """Generate a fresh base32 TOTP secret."""
    return pyotp.random_base32()


def totp_uri(secret, username):
    """otpauth:// provisioning URI for QR codes / manual entry."""
    return pyotp.TOTP(secret).provisioning_uri(name=username, issuer_name=MFA_ISSUER)


def verify_totp(secret, code):
    """Validate a 6-digit TOTP code, allowing +/-1 step of clock drift."""
    if not secret or not code:
        return False
    code = str(code).strip().replace(" ", "")
    if not code.isdigit():
        return False
    try:
        return pyotp.TOTP(secret).verify(code, valid_window=1)
    except Exception:
        return False


def generate_backup_codes(n=BACKUP_CODE_COUNT):
    """Return a list of human-friendly one-time codes (plaintext, shown once)."""
    codes = []
    for _ in range(n):
        raw = secrets.token_hex(5)  # 10 hex chars
        codes.append(f"{raw[:5]}-{raw[5:]}")
    return codes


def hash_backup_code(code):
    """Hash a backup code for storage (normalise case/separators first)."""
    norm = _normalise_backup(code)
    return generate_password_hash(norm)


def _normalise_backup(code):
    return (code or "").strip().lower().replace("-", "").replace(" ", "")


def verify_backup_code(user_id, code, db_path=db.DEFAULT_DB_PATH):
    """Check a backup code against the user's unused set; consume it on match.
    Returns True if accepted."""
    norm = _normalise_backup(code)
    if not norm:
        return False
    for code_id, code_hash in db.list_unused_backup_hashes(user_id, db_path=db_path):
        try:
            if check_password_hash(code_hash, norm):
                db.consume_backup_code(code_id, db_path=db_path)
                return True
        except Exception:
            continue
    return False


def enroll_backup_codes(user_id, db_path=db.DEFAULT_DB_PATH):
    """Generate, store (hashed), and return a fresh set of plaintext codes."""
    codes = generate_backup_codes()
    db.replace_backup_codes(user_id, [hash_backup_code(c) for c in codes],
                            db_path=db_path)
    return codes


def authenticate(username, password, db_path=db.DEFAULT_DB_PATH):
    """
    Verify credentials. Returns a session-user dict on success, else None.

    Resolution order:
      1. If user accounts exist -> validate against the users table.
      2. Else, fall back to the legacy NETSCAN_PASSWORD (any username), granting
         a transient 'admin' identity so first-time setup is possible.
    """
    username = (username or "").strip()
    if accounts_exist(db_path=db_path):
        user = db.get_user(username, db_path=db_path)
        if user and verify_password(user["password_hash"], password):
            # Note: caller is responsible for the MFA second factor before
            # establishing a session. We do NOT touch_login here so that
            # last_login reflects a fully-authenticated sign-in.
            return {"id": user["id"], "username": user["username"],
                    "role": user["role"],
                    "tenant": user.get("tenant") or "default",
                    "mfa_required": bool(user.get("mfa_required")),
                    "mfa_enrolled": user.get("mfa_secret") is not None}
        return None

    # Legacy fallback (no accounts yet).
    if LEGACY_PASSWORD and password == LEGACY_PASSWORD:
        return {"id": None, "username": username or "admin", "role": "admin",
                "tenant": "default", "legacy": True}
    return None


def auth_required():
    """Is authentication active at all? (accounts present OR legacy pw set)."""
    return accounts_exist() or bool(LEGACY_PASSWORD)