flowdeck/app/auth/session.py

"""FlowDeck — Session management with signed cookies."""
from __future__ import annotations

import logging
from datetime import UTC, datetime
from uuid import uuid4

from itsdangerous import BadSignature, SignatureExpired, URLSafeTimedSerializer

from app.config import settings

logger = logging.getLogger(__name__)

_serializer = URLSafeTimedSerializer(settings.app_secret_key)

# Colonnes de la table ``users`` qui ne doivent JAMAIS quitter le serveur :
# le cookie de session est signé (HMAC) mais pas chiffré — son payload est
# lisible en base64 par quiconque détient le cookie, et ``/auth/user`` le
# renvoyait tel quel au client.
_SECRET_USER_FIELDS = ("password_hash",)


def public_user(user_data: dict | None) -> dict | None:
    """Return a copy of ``user_data`` stripped of credential material."""
    if user_data is None:
        return None
    return {k: v for k, v in user_data.items() if k not in _SECRET_USER_FIELDS}


class SessionManager:
    """Manages user sessions via signed cookies (v5.2.0: revocable).

    Each cookie embeds a ``sid`` referencing a row in ``user_sessions``.
    Revoking that row instantly invalidates the cookie (checked in
    ``decode_session``). Legacy cookies without a ``sid`` stay valid.
    """

    @staticmethod
    def create_session(user_data: dict, request=None) -> str:
        """Create a signed session cookie value.

        ``request`` is optional — when provided the session is recorded in the
        ``user_sessions`` table (ip + user agent) and becomes revocable.
        """
        payload = {
            "user": public_user(user_data),
            "created_at": datetime.now(UTC).replace(tzinfo=None).isoformat(),
        }
        user_id = user_data.get("id")
        if user_id:
            sid = str(uuid4())
            payload["sid"] = sid
            _record_session(sid, user_id, request)
        return _serializer.dumps(payload)

    @staticmethod
    def decode_session(cookie: str) -> dict | None:
        """Decode and validate a session cookie. Returns user data or None."""
        try:
            payload = _serializer.loads(cookie, max_age=86400 * 7)  # 7 days
        except (BadSignature, SignatureExpired):
            return None

        sid = payload.get("sid") or ""
        if sid and not _session_active(sid):
            # Revoked or deleted session → treat as logged out.
            return None
        if sid:
            _touch_session(sid)
        return payload.get("user")

    @staticmethod
    def session_id(cookie: str) -> str | None:
        """Return the session id embedded in a cookie (or None)."""
        try:
            payload = _serializer.loads(cookie, max_age=86400 * 7)
            return payload.get("sid")
        except (BadSignature, SignatureExpired):
            return None

    @staticmethod
    def list_sessions(user_id: int) -> list[dict]:
        """All recorded sessions for a user (for the Settings UI)."""
        from app.db import get_conn
        with get_conn() as conn:
            rows = conn.execute(
                "SELECT id, ip_address, user_agent, created_at, last_seen_at, revoked "
                "FROM user_sessions WHERE user_id=? ORDER BY last_seen_at DESC",
                (user_id,),
            ).fetchall()
        return [dict(r) for r in rows]

    @staticmethod
    def revoke_session(sid: str) -> bool:
        """Revoke a session row. Returns True if a row was updated."""
        from app.db import get_conn
        with get_conn() as conn:
            cur = conn.execute(
                "UPDATE user_sessions SET revoked=1 WHERE id=? AND revoked=0", (sid,)
            )
            conn.commit()
            return cur.rowcount > 0

    @staticmethod
    def refresh_session(cookie: str, user_data: dict, request=None) -> str:
        """Re-sign a cookie keeping its session id (used after profile edits)."""
        sid = SessionManager.session_id(cookie) if cookie else None
        payload = {
            "user": public_user(user_data),
            "created_at": datetime.now(UTC).replace(tzinfo=None).isoformat(),
        }
        user_id = user_data.get("id")
        if user_id:
            if sid is None:
                sid = str(uuid4())
                _record_session(sid, user_id, request)
            payload["sid"] = sid
        return _serializer.dumps(payload)

    @staticmethod
    def store_token(user_id: int, gitea_token: str) -> None:
        """Store a user's Gitea OAuth token in SQLite."""
        from app.db import get_conn
        with get_conn() as conn:
            conn.execute(
                """INSERT INTO user_tokens (gitea_user_id, gitea_token, updated_at)
                   VALUES (?, ?, CURRENT_TIMESTAMP)
                   ON CONFLICT(gitea_user_id)
                   DO UPDATE SET gitea_token=excluded.gitea_token, updated_at=CURRENT_TIMESTAMP""",
                (user_id, gitea_token),
            )
            conn.commit()

    @staticmethod
    def get_token(user_id: int) -> str | None:
        """Get a user's stored Gitea token."""
        from app.db import get_conn
        with get_conn() as conn:
            row = conn.execute(
                "SELECT gitea_token FROM user_tokens WHERE gitea_user_id=?",
                (user_id,),
            ).fetchone()
        return row["gitea_token"] if row else None


def _record_session(sid: str, user_id: int, request) -> None:
    ip = ""
    ua = ""
    if request is not None:
        ip = request.client.host if getattr(request, "client", None) else ""
        ua = (request.headers.get("user-agent", "") or "")[:500]
    try:
        from app.db import get_conn
        with get_conn() as conn:
            conn.execute(
                "INSERT INTO user_sessions (id, user_id, ip_address, user_agent) VALUES (?, ?, ?, ?)",
                (sid, user_id, ip, ua),
            )
            conn.commit()
    except Exception as exc:  # table may not exist in very old installs
        logger.debug("session record skipped: %s", exc)


def _session_active(sid: str) -> bool:
    try:
        from app.db import get_conn
        with get_conn() as conn:
            row = conn.execute(
                "SELECT revoked FROM user_sessions WHERE id=?", (sid,)
            ).fetchone()
        return bool(row and not row["revoked"])
    except Exception:
        # No table / DB unavailable → keep the cookie valid (fail-open-safe).
        return True


def _touch_session(sid: str) -> None:
    try:
        from app.db import get_conn
        with get_conn() as conn:
            conn.execute(
                "UPDATE user_sessions SET last_seen_at=CURRENT_TIMESTAMP WHERE id=? AND revoked=0",
                (sid,),
            )
            conn.commit()
    except Exception:
        logger.exception("_touch_session")


# FastAPI dependency
def get_current_user(request) -> dict | None:
    """FastAPI dependency: extract current user from session cookie."""
    session = request.cookies.get("flowdeck_session")
    if session:
        return SessionManager.decode_session(session)
    return None