app/federation.py
Rama main · réplica de lectura del repositorio original.
"""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)