"""Authentication: proving who a caller is, and what that entitles them to.
One identity model, however many sign-in mechanisms you end up with. The
skeleton ships one - a password, kept in a signed session - and
`principal_for_user` is the seam every later one plugs into (an API token, an
OIDC or SAML callback, a passkey). Each of those needs a User row and an
organisation; none of them needs to care how the email was proved.
What is deliberately *not* here: an identity provider, and any notion of HTTP.
"""
import logging
import secrets
from datetime import datetime, timedelta
from werkzeug.security import check_password_hash, generate_password_hash
from db.db_declarations import Membership, Organization, User
from .context import ALL_SCOPES, Forbidden, Invalid, NotFound, Principal, Scope
logger = logging.getLogger(__name__)
# What each role may do. Stored as a role rather than as scopes so that
# tightening the meaning of "member" is one edit here instead of a migration over
# every membership row.
ROLE_SCOPES = {
"owner": ALL_SCOPES,
"admin": ALL_SCOPES,
"member": frozenset({Scope.READ, Scope.WRITE}),
"viewer": frozenset({Scope.READ}),
}
ROLES = tuple(ROLE_SCOPES)
# --- passwords --------------------------------------------------------------
MIN_PASSWORD_LENGTH = 12
def hash_password(password: str) -> str:
if len(password or "") < MIN_PASSWORD_LENGTH:
# Length is the only rule enforced. Composition rules push people toward
# predictable substitutions; a longer secret is worth more.
raise Invalid(f"A password needs to be at least {MIN_PASSWORD_LENGTH} characters.")
return generate_password_hash(password)
MAX_FAILED_LOGINS = 5
LOCKOUT = timedelta(minutes=15)
# A real hash of a value nobody will present, so the "no such user" path costs
# the same as the "wrong password" path. Computed once at import, not per call.
_DUMMY_HASH = generate_password_hash(secrets.token_urlsafe(32))
def authenticate_password(session, email: str, password: str) -> User:
"""Verify an email and password, or raise.
One error for every failure - unknown address, wrong password, disabled or
locked account - because distinguishing them turns the login form into an
account enumerator.
"""
email = (email or "").strip().lower()
user = session.query(User).filter(User.email == email).one_or_none()
now = datetime.utcnow()
locked = bool(user and user.locked_until and user.locked_until > now)
# Hash even when the user does not exist or is locked, so response time does
# not reveal which case this is.
stored = user.password_hash if user and user.password_hash else _DUMMY_HASH
ok = check_password_hash(stored, password or "")
if not user or not user.password_hash or not ok or not user.is_active or locked:
if user and not locked:
_record_failure(session, user, now)
raise Forbidden("That email and password do not match.")
user.failed_logins = 0
user.locked_until = None
user.last_login_at = now
session.commit()
return user
def _record_failure(session, user: User, now: datetime) -> None:
user.failed_logins = (user.failed_logins or 0) + 1
if user.failed_logins >= MAX_FAILED_LOGINS:
user.locked_until = now + LOCKOUT
logger.warning("Locked %s until %s after %d failed sign-ins.",
user.email, user.locked_until, user.failed_logins)
session.commit()
def change_password(session, principal: Principal, current: str, new: str) -> User:
"""Change the signed-in user's own password. The current password is required
even with a valid session - a session is not proof of knowing the password,
and without this check its holder could lock the owner out."""
principal.require(Scope.READ)
if principal.user_id is None:
raise Invalid("Only a signed-in account can change a password.")
user = session.get(User, principal.user_id)
if user is None:
raise NotFound("That account no longer exists.")
if not user.password_hash:
raise Invalid("This account signs in through an identity provider; there is "
"no password here to change.")
if not check_password_hash(user.password_hash, current or ""):
raise Forbidden("Your current password is not correct.")
if current == new:
raise Invalid("The new password is the same as the old one.")
hashed = hash_password(new) # enforces the length rule before anything is written
user.password_hash = hashed
user.failed_logins = 0
user.locked_until = None
session.commit()
return user
# --- resolving a caller ------------------------------------------------------
def memberships_for(session, user: User) -> list[tuple[Membership, Organization]]:
"""Every organisation this user belongs to, with the role held in each.
Ordered by organisation name so "the first one" is stable across requests."""
return (session.query(Membership, Organization)
.join(Organization, Membership.org_id == Organization.id)
.filter(Membership.user_id == user.id)
.order_by(Organization.name)
.all())
def principal_for_user(session, user: User, org_id: int | None = None,
surface: str = "ui") -> Principal:
"""The Principal for a user in one organisation - the seam every sign-in path
funnels through.
Scopes are derived here, on every request, rather than stored in the session,
so changing someone's role takes effect on their next click instead of on
their next sign-in.
"""
rows = memberships_for(session, user)
if not rows:
raise Forbidden("That account does not belong to any organisation yet.")
if org_id is None:
membership, _org = rows[0]
else:
membership = next((m for m, _ in rows if m.org_id == org_id), None)
if membership is None:
raise NotFound("No such organisation for this account.")
scopes = ROLE_SCOPES.get(membership.role)
if scopes is None:
logger.warning("Unknown membership role %r; granting nothing.", membership.role)
scopes = frozenset()
return Principal(subject=user.email, surface=surface, scopes=scopes,
org_id=membership.org_id, user_id=user.id)
# --- provisioning -----------------------------------------------------------
def _mint_account_id() -> str:
"""A stable, opaque public identifier for a user. 24 hex chars of CSPRNG
entropy: unguessable, and never an email, so it survives the user changing
address and leaks nothing when it appears in a URL or a payload."""
return secrets.token_hex(12)
def create_user(session, email: str, password: str | None,
name: str | None = None) -> User:
"""Create an account. `password` may be None for an account that will only
ever sign in through an identity provider - storing no hash is the honest
representation of that."""
email = (email or "").strip().lower()
if "@" not in email:
raise Invalid("That does not look like an email address.")
if session.query(User).filter(User.email == email).one_or_none():
raise Invalid(f"{email} already has an account.")
user = User(
email=email,
name=(name or "").strip() or None,
account_id=_mint_account_id(),
password_hash=hash_password(password) if password else None,
)
session.add(user)
session.commit()
return user
def add_membership(session, user: User, org: Organization,
role: str = "member") -> Membership:
"""Put a user in an organisation, or change the role they hold there."""
if role not in ROLE_SCOPES:
raise Invalid(f"Unknown role {role!r}. One of: {', '.join(ROLES)}.")
existing = (session.query(Membership)
.filter(Membership.user_id == user.id, Membership.org_id == org.id)
.one_or_none())
if existing:
existing.role = role
session.commit()
return existing
membership = Membership(user_id=user.id, org_id=org.id, role=role)
session.add(membership)
session.commit()
return membership