"""Signing in through somebody else's identity provider.

`appcore/oidc.py` does the protocol and knows nothing about this app. This file
is the other half: which account an identity belongs to, whether a new one may
be created, and what is remembered afterwards.

Two rules carry the security of the whole feature:

* **Accounts are matched on `(issuer, subject)`, never on email.** An address
  is something a directory can often be *told*, so matching on it would let
  anybody who can set an email in the provider take over an existing local
  account — decisions.md #119.
* **`provisioning` decides who may sign in at all.** `off` admits only linked
  accounts; `invite` admits a stranger holding a live invitation; `open`
  admits anyone the provider authenticates. The default is `off`, because the
  other two are statements about who a household's directory contains.
"""

from __future__ import annotations

import datetime as dt
import hashlib
import secrets
import time

from sqlalchemy import delete, select
from sqlalchemy.orm import Session as DbSession

from appcore import ensure_utc, oidc

from . import twofactor
from .models import ExternalIdentity, OidcState, PortfolioInvite, User, UserSession
from .settings import OidcSettings, PortfolioSettings

# Long enough for a slow provider and a password prompt, short enough that an
# abandoned attempt is not a live row for the afternoon.
STATE_TTL = dt.timedelta(minutes=15)


class FederationError(Exception):
    """A sign-in that cannot be completed. The message is safe to show."""


def _utcnow() -> dt.datetime:
    return dt.datetime.now(dt.timezone.utc)


# What a provider says is input too, and its columns have limits. An id or an
# email cut short is a different id or email, so neither is cut: one that cannot
# be stored is refused where it identifies somebody, and dropped where it only
# describes them. A name is the app's to fit (decisions.md #131). A NUL makes any
# of them unusable: Postgres refuses it even in a lookup.

def storable(value: str | None, column) -> bool:
    return value is not None and "\x00" not in value and len(value) <= column.type.length


def _require_storable_subject(identity: oidc.Identity) -> None:
    if not storable(identity.subject, ExternalIdentity.subject):
        raise FederationError(
            "Your identity provider sent an account id this app cannot store.")


def session_id(identity: oidc.Identity) -> str | None:
    """The provider's session id, kept for back-channel logout, or None when it
    cannot be stored. Without it that one session ends only here."""
    sid = identity.session_id
    return sid if storable(sid, UserSession.oidc_sid) else None


def _hash(raw: str) -> str:
    return hashlib.sha256(raw.encode()).hexdigest()


# --------------------------------------------------------------------------- #
# Whether it can be offered at all
# --------------------------------------------------------------------------- #

def _is_public(conf: OidcSettings) -> bool:
    """A public client has no secret to be missing. Only the exact value
    counts, so a typo leaves the secret required — decisions.md #121."""
    return conf.client_auth == "none"


def configured(settings: PortfolioSettings) -> bool:
    conf = settings.auth.oidc
    return bool(conf.enabled and conf.issuer and conf.client_id
                and (conf.client_secret or _is_public(conf))
                and conf.redirect_uri)


def unavailable_reason(settings: PortfolioSettings) -> str | None:
    """Why sign-in cannot be offered, in a sentence, or None."""
    conf = settings.auth.oidc
    if not conf.enabled:
        return "Turn on single sign-on in Admin → Settings."
    if not conf.issuer or not conf.client_id:
        return "Set the provider URL and client ID in Admin → Settings."
    if not conf.client_secret and not _is_public(conf):
        # Both ways out: a provider that issues public clients has no secret
        # to fetch, and `client_auth` is the setting they would have no reason
        # to have found yet — raised by review on #25.
        return ("Set auth.oidc.client_secret in your configuration file, or "
                "auth.oidc.client_auth to 'none' for a public client.")
    if not conf.redirect_uri:
        return "Set the redirect URL in Admin → Settings."
    return None


# How long what a provider's discovery said is kept: an hour, so an endpoint
# it moves is picked up without a restart.
DISCOVERY_SECONDS = 3600

_provider: tuple[tuple, float, oidc.Provider] | None = None


def provider(settings: PortfolioSettings) -> oidc.Provider:
    """The configured provider, one per configuration for up to an hour.
    appcore keeps what discovery said on the instance, and a new one per call
    fetched it again every time.
    A changed setting is a different configuration, so it takes effect without
    a restart."""
    global _provider
    conf = settings.auth.oidc
    key = (conf.issuer or "", conf.client_id or "", conf.client_secret,
           conf.redirect_uri or "", tuple(conf.scopes), conf.client_auth)
    now = time.monotonic()
    if _provider is None or _provider[0] != key or now - _provider[1] > DISCOVERY_SECONDS:
        _provider = (key, now, oidc.Provider(
            issuer=key[0], client_id=key[1], client_secret=key[2],
            redirect_uri=key[3], scopes=key[4], client_auth=key[5]))
    return _provider[2]


# --------------------------------------------------------------------------- #
# The redirect out
# --------------------------------------------------------------------------- #

def begin(db: DbSession, settings: PortfolioSettings,
          invite: PortfolioInvite | None = None,
          link_to: User | None = None) -> tuple[str, str]:
    """Where to send the browser, and the token that remembers this attempt.

    `link_to` marks it as attaching a provider to an account that already
    exists, rather than signing one in.
    """
    if not configured(settings):
        raise FederationError(unavailable_reason(settings) or "Not available.")
    url, pending = oidc.begin(provider(settings))
    raw = secrets.token_urlsafe(32)
    db.add(OidcState(
        token_hash=_hash(raw),
        state=pending["state"],
        nonce=pending["nonce"],
        verifier=pending["verifier"],
        invite_id=invite.id if invite else None,
        link_user_id=link_to.id if link_to else None,
        expires_at=_utcnow() + STATE_TTL,
    ))
    db.flush()
    return url, raw


def take_state(db: DbSession, raw: str | None, state: str) -> OidcState:
    """The attempt behind this token, consumed as it is read.

    The `state` the provider echoed must match the one issued: that is what
    stops a callback somebody else constructed from completing a sign-in here.
    """
    if not raw:
        raise FederationError("That sign-in attempt has expired. Try again.")
    row = db.scalar(select(OidcState).where(OidcState.token_hash == _hash(raw)))
    if row is None or ensure_utc(row.expires_at) < _utcnow():
        if row is not None:
            db.delete(row)
            db.flush()
        raise FederationError("That sign-in attempt has expired. Try again.")
    # Compared before the row is spent, so a mismatch cannot be used to burn
    # somebody else's attempt.
    if not secrets.compare_digest(row.state, state or ""):
        raise FederationError("That sign-in did not match the request that started it.")
    return row


def purge_expired_states(db: DbSession) -> int:
    """Attempts nobody came back from. Runs with maintenance."""
    result = db.execute(delete(OidcState).where(OidcState.expires_at < _utcnow()))
    return result.rowcount or 0


# --------------------------------------------------------------------------- #
# The callback
# --------------------------------------------------------------------------- #

def resolve(db: DbSession, settings: PortfolioSettings,
            identity: oidc.Identity, invite: PortfolioInvite | None) -> User:
    """The account this identity signs in as, creating one only if allowed.

    Never matched on email. A local account with the same address is a
    different thing until somebody links them deliberately.
    """
    _require_storable_subject(identity)
    known = db.scalar(
        select(ExternalIdentity).where(
            ExternalIdentity.issuer == identity.issuer,
            ExternalIdentity.subject == identity.subject,
        )
    )
    if known is not None:
        known.last_used_at = _utcnow()
        # Describes them, so one that cannot be stored is not a reason to
        # refuse the sign-in: the one already on file stays.
        if identity.email is None or storable(identity.email, ExternalIdentity.email):
            known.email = identity.email
        db.flush()
        user = db.get(User, known.user_id)
        if user is None or not user.is_active:
            raise FederationError("That account is no longer active.")
        return user

    mode = settings.auth.oidc.provisioning
    if mode == "invite" and invite is None:
        raise FederationError(
            "This app does not know you yet. Ask for an invitation link.")
    if mode not in ("invite", "open"):
        raise FederationError(
            "This app does not know you yet. An administrator has to link your "
            "account first.")
    return provision(db, identity)


def provision(db: DbSession, identity: oidc.Identity) -> User:
    """Create an account for an identity nobody has seen before.

    The email is stored because an account needs one to be addressed by, and
    an account with a colliding address is refused rather than joined: two
    people are not one person because a directory says so.
    """
    email = (identity.email or f"{identity.subject}@{identity.issuer}").lower()
    # Measured lowercased, as it is stored: some characters grow (#131).
    if not storable(email, User.email):
        raise FederationError(
            "Your identity provider sent an email address this app cannot store.")
    if db.scalar(select(User).where(User.email == email)) is not None:
        raise FederationError(
            "An account already uses that email address. Sign in with it, then "
            "link your provider from your profile.")
    # No password: `auth.verify_password` refuses a NULL hash, so this account
    # signs in one way until somebody sets one — decisions.md #120.
    name = (identity.name or "").replace("\x00", "").strip() or email
    user = User(email=email, name=name[:User.name.type.length], password_hash=None)
    db.add(user)
    db.flush()
    # Issued here as on every other path that creates an account. It matters
    # more here than anywhere: this account has no password either, so without
    # them a provider that goes away leaves nothing but an admin reset.
    # `recovery_codes_seen_at` stays NULL, so the banner asks them to save a
    # set of their own on first sign-in.
    twofactor.ensure_recovery_codes(db, user)
    link(db, user, identity)
    return user


def link(db: DbSession, user: User, identity: oidc.Identity) -> ExternalIdentity:
    """Attach an identity to an account. Refuses one already attached
    elsewhere — a subject belongs to exactly one account."""
    _require_storable_subject(identity)
    existing = db.scalar(
        select(ExternalIdentity).where(
            ExternalIdentity.issuer == identity.issuer,
            ExternalIdentity.subject == identity.subject,
        )
    )
    if existing is not None and existing.user_id != user.id:
        raise FederationError("That provider account is already linked to "
                              "somebody else here.")
    if existing is not None:
        return existing
    email = identity.email if storable(identity.email, ExternalIdentity.email) else None
    row = ExternalIdentity(user_id=user.id, issuer=identity.issuer,
                           subject=identity.subject, email=email,
                           last_used_at=_utcnow())
    db.add(row)
    db.flush()
    return row


def identities_for(db: DbSession, user: User) -> list[ExternalIdentity]:
    return list(db.scalars(
        select(ExternalIdentity)
        .where(ExternalIdentity.user_id == user.id)
        .order_by(ExternalIdentity.created_at)
    ))


def unlink(db: DbSession, user: User, identity_id: int) -> bool:
    """Detach one provider from this account.

    Refused when it is the only way in: an account with no password and no
    other identity would be unreachable, and "you can no longer sign in" is
    not something a settings page should be able to do quietly.
    """
    rows = identities_for(db, user)
    row = next((r for r in rows if r.id == identity_id), None)
    if row is None:
        return False
    if user.password_hash is None and len(rows) == 1:
        raise FederationError(
            "Set a password first — this is the only way into your account.")
    db.delete(row)
    db.flush()
    return True


def end_sessions_for(db: DbSession, notice: oidc.LogoutNotice) -> int:
    """End the sessions a verified logout notice names. Returns how many.

    **Sessions only.** `is_active` is deliberately untouched: letting a
    provider disable an account is a far larger authority than ending a
    session, and it would break the promise that local accounts keep working
    when the provider is down. Somebody holding a password signs straight back
    in, and that is the intended behaviour — decisions.md #127.

    Two keys, in order of precision:

    * `sid` — exactly the session the provider meant. One browser signing out
      there must not sign the person out of every other one.
    * `sub` — every session for that identity, which is more than the provider
      asked for but all that can be done with a token carrying only a subject.

    Both are scoped to sessions that actually began at the provider
    (`via_oidc`). A password sign-in has no provider session, so nothing the
    provider says about one can reach it.
    """
    from .models import UserSession

    if notice.session_id:
        # No `via_oidc` filter here, deliberately: `oidc_sid` is written in
        # exactly one place and always alongside `via_oidc`, so a local session
        # holds NULL and cannot match a sid at all. Adding the clause looked
        # like defence in depth and was unreachable — a guard no test can fail
        # is one that implies a threat that does not exist.
        # `test_only_a_provider_session_ever_carries_a_sid` pins the invariant
        # that makes this safe.
        rows = list(db.scalars(
            select(UserSession).where(UserSession.oidc_sid == notice.session_id)
        ))
    elif notice.subject:
        # The identities this subject holds HERE — matched on (issuer, subject)
        # and never on email, the same rule `resolve` follows.
        user_ids = list(db.scalars(
            select(ExternalIdentity.user_id).where(
                ExternalIdentity.issuer == notice.issuer,
                ExternalIdentity.subject == notice.subject,
            )
        ))
        rows = list(db.scalars(
            select(UserSession).where(
                UserSession.user_id.in_(user_ids or [-1]),
                UserSession.via_oidc.is_(True),
            )
        )) if user_ids else []
    else:  # pragma: no cover - appcore refuses a token naming neither
        rows = []

    for row in rows:
        db.delete(row)
    db.flush()
    return len(rows)