"""Login via Nextcloud's Login Flow v2 (the same mechanism NC's own desktop/ mobile clients use to obtain an app password without ever seeing a third-party app's code touch the user's real password). Flow, doc-verified against docs.nextcloud.com/server/stable/developer_manual /client_apis/LoginFlow/index.html: 1. POST {NC_BASE_URL}/index.php/login/v2 (no auth) -> {poll: {token, endpoint}, login: } 2. User opens `login`, authenticates directly with Nextcloud, approves. 3. POST {poll.endpoint} with body token= -> 404 while pending, 200 once (single-use) with {server, loginName, appPassword} once granted. Token valid 20 minutes. Because our backend does the polling (not browser JS), none of the CORS issues that rule out a pure-frontend NC login apply here - the browser only ever does a full-page navigation to NC's own login page and back. """ from __future__ import annotations import hashlib import secrets import time from datetime import datetime, timedelta from functools import wraps import requests from cryptography.fernet import Fernet, InvalidToken from flask import g, jsonify, request from .config import Config from .db import get_conn # In-memory store for login flows in progress: flow_id -> {poll_token, # poll_endpoint, created_at}. Short-lived (20 min NC-side) and low-stakes if # lost on a backend restart (the user just retries the login button) - not # worth a DB table. _pending_flows: dict[str, dict] = {} _FLOW_TTL_SECONDS = 20 * 60 def _fernet() -> Fernet: if not Config.TOKEN_ENCRYPTION_KEY: raise RuntimeError( "TOKEN_ENCRYPTION_KEY is not set (see .env.example) - refusing to " "store or read an NC app password without it" ) return Fernet(Config.TOKEN_ENCRYPTION_KEY.encode()) def _hash_session_id(session_id: str) -> str: return hashlib.sha256(session_id.encode()).hexdigest() # ------------------------------------------------------------ login flow def start_login_flow() -> dict: """Kicks off a Login Flow v2 against NC. Returns {flow_id, login_url}.""" resp = requests.post(f"{Config.NC_BASE_URL}/index.php/login/v2", timeout=15) resp.raise_for_status() data = resp.json() flow_id = secrets.token_urlsafe(24) _pending_flows[flow_id] = { "poll_token": data["poll"]["token"], "poll_endpoint": data["poll"]["endpoint"], "created_at": time.time(), } _gc_expired_flows() return {"flow_id": flow_id, "login_url": data["login"]} def poll_login_flow(flow_id: str) -> dict: """Returns {"status": "pending"} | {"status": "expired"} | {"status": "done", "nc_user_id": str, "session_id": str}. On success, provisions/updates the user row and creates a new session - the caller still has to actually set the cookie on the response. """ flow = _pending_flows.get(flow_id) if flow is None: return {"status": "expired"} if time.time() - flow["created_at"] > _FLOW_TTL_SECONDS: del _pending_flows[flow_id] return {"status": "expired"} resp = requests.post( flow["poll_endpoint"], data={"token": flow["poll_token"]}, timeout=15 ) if resp.status_code == 404: return {"status": "pending"} resp.raise_for_status() # Single-use - NC only returns this once, so this flow is done either way. del _pending_flows[flow_id] granted = resp.json() nc_user_id = granted["loginName"] app_password = granted["appPassword"] encrypted = _fernet().encrypt(app_password.encode()) with get_conn() as conn: conn.execute( """INSERT INTO users (nc_user_id, nc_app_password_encrypted) VALUES (?, ?) ON CONFLICT(nc_user_id) DO UPDATE SET nc_app_password_encrypted = excluded.nc_app_password_encrypted, last_login_at = datetime('now')""", (nc_user_id, encrypted), ) session_id = create_session(nc_user_id) return {"status": "done", "nc_user_id": nc_user_id, "session_id": session_id} def _gc_expired_flows() -> None: now = time.time() expired = [fid for fid, f in _pending_flows.items() if now - f["created_at"] > _FLOW_TTL_SECONDS] for fid in expired: del _pending_flows[fid] # --------------------------------------------------------------- sessions def create_session(nc_user_id: str) -> str: """Returns the raw session id to set as the cookie value.""" session_id = secrets.token_urlsafe(32) expires_at = (datetime.utcnow() + timedelta(days=Config.SESSION_TTL_DAYS)).isoformat() with get_conn() as conn: conn.execute( "INSERT INTO sessions (session_id_hash, nc_user_id, expires_at) VALUES (?, ?, ?)", (_hash_session_id(session_id), nc_user_id, expires_at), ) return session_id def resolve_session(session_id: str) -> str | None: """Returns the nc_user_id for a valid, non-expired session, refreshing its expiry (sliding session), or None if invalid/expired.""" session_hash = _hash_session_id(session_id) with get_conn() as conn: row = conn.execute( "SELECT nc_user_id, expires_at FROM sessions WHERE session_id_hash = ?", (session_hash,), ).fetchone() if row is None: return None if datetime.fromisoformat(row["expires_at"]) < datetime.utcnow(): conn.execute("DELETE FROM sessions WHERE session_id_hash = ?", (session_hash,)) return None new_expiry = (datetime.utcnow() + timedelta(days=Config.SESSION_TTL_DAYS)).isoformat() conn.execute( "UPDATE sessions SET expires_at = ? WHERE session_id_hash = ?", (new_expiry, session_hash), ) return row["nc_user_id"] def destroy_session(session_id: str) -> None: with get_conn() as conn: conn.execute( "DELETE FROM sessions WHERE session_id_hash = ?", (_hash_session_id(session_id),) ) def get_app_password(nc_user_id: str) -> str: with get_conn() as conn: row = conn.execute( "SELECT nc_app_password_encrypted FROM users WHERE nc_user_id = ?", (nc_user_id,) ).fetchone() if row is None: raise RuntimeError(f"no stored credentials for {nc_user_id!r}") try: return _fernet().decrypt(bytes(row["nc_app_password_encrypted"])).decode() except InvalidToken as exc: raise RuntimeError( "stored app password could not be decrypted - TOKEN_ENCRYPTION_KEY " "changed or is wrong" ) from exc # ------------------------------------------------------------- decorator def enforce_login(): """Populates g.nc_user_id/g.nc_app_password from the session cookie, or returns a 401 response if there isn't a valid one. Used both as a blueprint-wide `before_request` (so any route added to that blueprint is protected by default, not only ones someone remembered to decorate) and directly by the `login_required` decorator for routes registered elsewhere. """ session_id = request.cookies.get(Config.SESSION_COOKIE_NAME) nc_user_id = resolve_session(session_id) if session_id else None if nc_user_id is None: return jsonify(error="not logged in"), 401 g.nc_user_id = nc_user_id g.nc_app_password = get_app_password(nc_user_id) return None def login_required(view): """Per-route variant of enforce_login(), for routes on a blueprint that isn't (or shouldn't be) guarded wholesale.""" @wraps(view) def wrapped(*args, **kwargs): rejection = enforce_login() if rejection is not None: return rejection return view(*args, **kwargs) return wrapped