knacky/Metamorph
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(m2): auth, JWT, invitations, bootstrap, RTOps SPA pages

Browse Source 
Crypto + tokens
- app/core/security.py: Argon2id PasswordHasher (time_cost=2, memory_cost=
  64 MiB, parallelism=2) + opaque-token SHA-256 helpers (raw token shown
  once, only the hash lives in the DB).
- app/core/jwt_tokens.py: HS256, claims iss/sub/type/jti/iat/exp. Access
  1h, refresh 30d.

Services
- services/auth.py: login, refresh with token rotation + reuse-detection
  chain revoke, logout (idempotent), change_password (forces logout-all).
- services/invitations.py: create, preview, accept, revoke. Default 7d TTL.
- services/bootstrap.py: seeds the 3 system groups (admin/redteam/blueteam),
  consumes the install token, attaches the first user to admin.
- core/install_token.py: mints, persists in settings, marks consumed,
  regenerate hook for /diag/reset.

API
- POST /setup (consume install token, create 1st admin) + GET /setup
  (status).
- POST /auth/{login,refresh,logout,change-password} + GET /auth/me.
- POST /invitations + GET /invitations + GET /invitations/preview/<token> +
  POST /invitations/accept/<token> + POST /invitations/<id>/revoke.
- POST /diag/reset: test-only kill switch (truncate auth tables + mint
  fresh install token). Allowed in dev too (with WARNING log) so the e2e
  suite can run against a make-up stack; production locked out.

Middleware
- @require_auth populates g.current_user (snapshot dataclass, session
  closed before request handler runs).
- @require_perm(*codes): atomic perm union check; admin group bypasses.
  Perm catalogue lands in M3, scaffolding here.
- flask-limiter: 10/min/IP on /auth/login & /auth/refresh, 5/min on
  /auth/change-password & /setup, 10–20/min on invitation endpoints.
  Disabled in APP_ENV=test.

CLI
- flask --app app.cli metamorph print-install-token [--force]
- flask --app app.cli metamorph seed-mitre (M4 placeholder)

Refresh cookie metamorph_refresh: HttpOnly + Secure (localhost is a secure
context for modern browsers) + SameSite=Strict + Path=/api/v1/auth/.

Email validation: app.api._validation.Email permissive RFC-shape regex so
internal TLDs (.local/.corp/.test) are accepted — pydantic.EmailStr's
deliverability check is too strict for red-team labs.

Frontend
- lib/{api,auth}.ts: access token in module memory, refresh cookie,
  automatic 401-retry via /auth/refresh, useAuth() hook.
- components/{Layout,RequireAuth}.tsx + ui/{TextField,Alert}.tsx.
- pages/{Login,Setup,Register,Profile}.

Testing
- tests/test_auth_flow.py: 15 integration tests (24 backend total).
- e2e/tests/m2-auth.spec.ts: 8 Playwright tests (20 e2e total).
- tasks/testing-m2.md.

DoD: make test-api → 24 passed, make e2e → 20 passed; spec-reviewer pass
applied (Secure unconditional, refresh limit 10/min/IP).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Knacky
2026-05-11 06:16:48 +02:00
parent e995853f0d
commit 700b563297
27 changed files with 3123 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
backend/app/api/_validation.py Normal file
View File

@@ -0,0 +1,30 @@
"""Lightweight email validator that tolerates internal/lab TLDs (.local, .corp, …).
`pydantic.EmailStr` relies on `email-validator` with `globally_deliverable=True`,
which rejects RFC 6761 special-use domains. Red-team and corporate intranet
deployments routinely use such suffixes — we accept any RFC-shape email and
defer deliverability checks to the operator.
"""
from __future__ import annotations
import re
from typing import Annotated
from pydantic import AfterValidator
# Permissive RFC-shape pattern: local-part 1..64 chars, domain has at least one
# dot, each label is 1..63 chars of letters/digits/hyphens, total ≤ 254.
_EMAIL_RE = re.compile(
r"^(?=.{1,254}$)[A-Za-z0-9._%+\-]{1,64}@[A-Za-z0-9](?:[A-Za-z0-9\-]{0,61}[A-Za-z0-9])?(?:\.[A-Za-z0-9](?:[A-Za-z0-9\-]{0,61}[A-Za-z0-9])?)+$"
)
def _validate_email(value: str) -> str:
v = value.strip()
if not _EMAIL_RE.match(v):
raise ValueError("not a valid email address")
return v.lower()
Email = Annotated[str, AfterValidator(_validate_email)]

157
backend/app/api/auth.py Normal file
View File

@@ -0,0 +1,157 @@
"""Authentication endpoints.
`POST /auth/login` returns the access token in the body and sets the refresh
token in an HTTPOnly cookie scoped to `/api/v1/auth/`. The cookie is
`Secure; SameSite=Strict` and only the matching paths can read it.
"""
from __future__ import annotations
import logging
from flask import Blueprint, g, jsonify, make_response, request
from pydantic import BaseModel, Field, ValidationError
from app.api._validation import Email
from app.core.auth_decorators import require_auth
from app.core.config import settings
from app.core.rate_limit import limiter
from app.services import auth as auth_svc
bp = Blueprint("auth", __name__, url_prefix="/auth")
log = logging.getLogger("metamorph.api.auth")
REFRESH_COOKIE_NAME = "metamorph_refresh"
REFRESH_COOKIE_PATH = "/api/v1/auth/"
class LoginPayload(BaseModel):
email: Email
password: str = Field(min_length=1)
class ChangePasswordPayload(BaseModel):
current_password: str = Field(min_length=1)
new_password: str = Field(min_length=8)
def _set_refresh_cookie(resp, token: str, expires_at) -> None:
resp.set_cookie(
REFRESH_COOKIE_NAME,
token,
expires=expires_at,
httponly=True,
secure=True, # spec §M2; localhost is a secure context for modern browsers
samesite="Strict",
path=REFRESH_COOKIE_PATH,
)
def _clear_refresh_cookie(resp) -> None:
resp.set_cookie(
REFRESH_COOKIE_NAME,
"",
expires=0,
httponly=True,
secure=True, # spec §M2; localhost is a secure context for modern browsers
samesite="Strict",
path=REFRESH_COOKIE_PATH,
)
def _read_refresh_cookie() -> str | None:
return request.cookies.get(REFRESH_COOKIE_NAME)
def _serialize_pair(pair: auth_svc.TokenPair) -> dict:
return {
"access_token": pair.access_token,
"token_type": "Bearer",
"user_id": str(pair.user_id),
}
@bp.post("/login")
@limiter.limit("10 per minute")
def login():
try:
payload = LoginPayload.model_validate(request.get_json(silent=True) or {})
except ValidationError as e:
return jsonify({"error": "invalid_request", "details": e.errors()}), 400
try:
pair = auth_svc.login(payload.email, payload.password)
except auth_svc.InvalidCredentials:
return jsonify({"error": "invalid_credentials"}), 401
resp = make_response(jsonify(_serialize_pair(pair)))
_set_refresh_cookie(resp, pair.refresh_token, pair.refresh_expires_at)
return resp
@bp.post("/refresh")
@limiter.limit("10 per minute")
def refresh_endpoint():
raw = _read_refresh_cookie()
if not raw:
return jsonify({"error": "no_refresh_cookie"}), 401
try:
pair = auth_svc.refresh(raw)
except auth_svc.TokenRevoked:
resp = make_response(jsonify({"error": "token_revoked"}), 401)
_clear_refresh_cookie(resp)
return resp
except auth_svc.InvalidCredentials:
resp = make_response(jsonify({"error": "invalid_refresh"}), 401)
_clear_refresh_cookie(resp)
return resp
resp = make_response(jsonify(_serialize_pair(pair)))
_set_refresh_cookie(resp, pair.refresh_token, pair.refresh_expires_at)
return resp
@bp.post("/logout")
def logout():
raw = _read_refresh_cookie()
if raw:
auth_svc.logout(raw)
resp = make_response(jsonify({"ok": True}))
_clear_refresh_cookie(resp)
return resp
@bp.get("/me")
@require_auth
def me():
u = g.current_user
return jsonify(
{
"id": str(u.id),
"email": u.email,
"display_name": u.display_name,
"locale": u.locale,
"is_admin": u.is_admin,
"groups": sorted(u.group_names),
"permissions": sorted(u.permissions),
}
)
@bp.post("/change-password")
@require_auth
@limiter.limit("5 per minute")
def change_password():
try:
payload = ChangePasswordPayload.model_validate(request.get_json(silent=True) or {})
except ValidationError as e:
return jsonify({"error": "invalid_request", "details": e.errors()}), 400
try:
auth_svc.change_password(g.current_user.id, payload.current_password, payload.new_password)
except auth_svc.InvalidCredentials:
return jsonify({"error": "current_password_incorrect"}), 400
except ValueError as e:
return jsonify({"error": "weak_password", "message": str(e)}), 400
resp = make_response(jsonify({"ok": True}))
_clear_refresh_cookie(resp)
return resp

146
backend/app/api/invitations.py Normal file
View File

@@ -0,0 +1,146 @@
"""Invitation endpoints — admin issues, invitee previews + accepts."""
from __future__ import annotations
import logging
import uuid
from flask import Blueprint, g, jsonify, make_response, request
from pydantic import BaseModel, Field, ValidationError
from app.api._validation import Email
from app.core.auth_decorators import require_auth, require_perm
from app.core.rate_limit import limiter
from app.services import invitations as inv_svc
bp = Blueprint("invitations", __name__, url_prefix="/invitations")
log = logging.getLogger("metamorph.api.invitations")
class CreateInvitationPayload(BaseModel):
email_hint: Email | None = None
group_ids: list[uuid.UUID] = Field(default_factory=list)
ttl_days: int | None = Field(default=None, ge=1, le=30)
class AcceptInvitationPayload(BaseModel):
email: Email
password: str = Field(min_length=8)
display_name: str | None = None
@bp.post("")
@require_auth
@require_perm("invitation.create")
def create():
try:
payload = CreateInvitationPayload.model_validate(request.get_json(silent=True) or {})
except ValidationError as e:
return jsonify({"error": "invalid_request", "details": e.errors()}), 400
from datetime import timedelta
ttl = (
timedelta(days=payload.ttl_days)
if payload.ttl_days is not None
else inv_svc.INVITATION_TTL
)
result = inv_svc.create_invitation(
created_by_user_id=g.current_user.id,
email_hint=payload.email_hint,
group_ids=payload.group_ids,
ttl=ttl,
)
log.info(
"metamorph.invitation.created",
extra={
"invitation_id": str(result.invitation_id),
"by_user_id": str(g.current_user.id),
"expires_at": result.expires_at.isoformat(),
},
)
return make_response(
jsonify(
{
"id": str(result.invitation_id),
"token": result.raw_token, # shown ONCE
"expires_at": result.expires_at.isoformat(),
}
),
201,
)
@bp.get("")
@require_auth
@require_perm("invitation.read")
def list_active():
rows = inv_svc.list_active()
return jsonify(
[
{
"id": str(r.id),
"email_hint": r.email_hint,
"expires_at": r.expires_at.isoformat(),
"groups": [g.name for g in r.pre_assigned_groups],
}
for r in rows
]
)
@bp.post("/<invitation_id>/revoke")
@require_auth
@require_perm("invitation.revoke")
def revoke(invitation_id: str):
try:
iid = uuid.UUID(invitation_id)
except ValueError:
return jsonify({"error": "invalid_id"}), 400
ok = inv_svc.revoke(iid)
if not ok:
return jsonify({"error": "not_revocable"}), 404
return jsonify({"ok": True})
@bp.get("/preview/<token>")
@limiter.limit("20 per minute")
def preview(token: str):
p = inv_svc.preview(token)
return jsonify(
{
"is_valid": p.is_valid,
"reason": p.reason,
"email_hint": p.email_hint,
"expires_at": p.expires_at.isoformat(),
"groups": p.groups,
}
)
@bp.post("/accept/<token>")
@limiter.limit("10 per minute")
def accept(token: str):
try:
payload = AcceptInvitationPayload.model_validate(request.get_json(silent=True) or {})
except ValidationError as e:
return jsonify({"error": "invalid_request", "details": e.errors()}), 400
try:
user_id = inv_svc.accept(
token,
email=payload.email,
password=payload.password,
display_name=payload.display_name,
)
except inv_svc.InvitationExpired:
return jsonify({"error": "invitation_expired"}), 410
except inv_svc.InvitationConsumed:
return jsonify({"error": "invitation_consumed"}), 410
except inv_svc.InvitationRevoked:
return jsonify({"error": "invitation_revoked"}), 410
except inv_svc.InvitationError as e:
return jsonify({"error": "invitation_invalid", "message": str(e)}), 400
except ValueError as e:
return jsonify({"error": "invalid_request", "message": str(e)}), 400
return make_response(jsonify({"ok": True, "user_id": str(user_id)}), 201)

79
backend/app/api/setup.py Normal file
View File

@@ -0,0 +1,79 @@
"""Bootstrap endpoint — consumes the install token to create the first admin."""
from __future__ import annotations
import logging
from flask import Blueprint, jsonify, make_response, request
from pydantic import BaseModel, Field, ValidationError
from app.api._validation import Email
from sqlalchemy import select
from app.core.rate_limit import limiter
from app.db.session import session_scope
from app.models.auth import User
from app.services.bootstrap import (
BootstrapError,
bootstrap_admin,
ensure_system_groups,
)
bp = Blueprint("setup", __name__, url_prefix="/setup")
log = logging.getLogger("metamorph.api.setup")
class SetupPayload(BaseModel):
install_token: str = Field(min_length=20)
email: Email
password: str = Field(min_length=8)
display_name: str | None = None
@bp.get("")
def setup_status():
"""Tell the SPA whether the bootstrap has already been done.
Used by the front to redirect to /setup vs /login on first paint.
"""
with session_scope() as s:
any_user = s.scalar(select(User.id).limit(1)) is not None
return jsonify({"completed": any_user})
@bp.post("")
@limiter.limit("5 per minute")
def setup():
try:
payload = SetupPayload.model_validate(request.get_json(silent=True) or {})
except ValidationError as e:
return jsonify({"error": "invalid_request", "details": e.errors()}), 400
try:
result = bootstrap_admin(
install_token=payload.install_token,
email=payload.email,
password=payload.password,
display_name=payload.display_name,
)
except BootstrapError as e:
return jsonify({"error": "bootstrap_failed", "message": str(e)}), 409
except ValueError as e:
return jsonify({"error": "invalid_request", "message": str(e)}), 400
log.warning(
"metamorph.bootstrap.completed",
extra={"user_id": str(result.user_id), "admin_group_id": str(result.admin_group_id)},
)
# Make sure the redteam/blueteam groups exist too (idempotent).
ensure_system_groups()
return make_response(
jsonify(
{
"ok": True,
"user_id": str(result.user_id),
}
),
201,
)

65
backend/app/cli.py Normal file
View File

@@ -0,0 +1,65 @@
"""Flask CLI entry point.
Used as `flask --app app.cli metamorph <subcommand>` (or via the make targets).
"""
from __future__ import annotations
import sys
import click
from flask import Flask
from flask.cli import AppGroup
from app.core.install_token import (
ensure_install_token,
log_install_token_banner,
regenerate_install_token,
)
from app.core.logging import configure_logging
from app.services.bootstrap import ensure_system_groups
from app.core.config import settings
def _create_cli_app() -> Flask:
configure_logging(settings.LOG_LEVEL)
return Flask("metamorph-cli")
app = _create_cli_app()
metamorph = AppGroup("metamorph", help="Metamorph admin commands.")
@metamorph.command("print-install-token")
@click.option(
"--force",
is_flag=True,
help="Always mint a fresh token even if one is already pending.",
)
def print_install_token(force: bool):
"""Mint and print the bootstrap install token (idempotent unless --force)."""
ensure_system_groups()
if force:
token = regenerate_install_token()
else:
token = ensure_install_token()
if token is None:
click.echo(
"No install token minted: either at least one user already exists, "
"or a token is already pending (use --force to mint a fresh one).",
err=True,
)
sys.exit(1)
log_install_token_banner(token)
@metamorph.command("seed-mitre")
def seed_mitre():
"""Placeholder for M4 — left so `make seed-mitre` doesn't crash."""
click.echo("MITRE seeding will land in M4. (no-op for now)", err=True)
sys.exit(0)
app.cli.add_command(metamorph)

139
backend/app/core/auth_decorators.py Normal file
View File

@@ -0,0 +1,139 @@
"""Flask decorators for authentication + authorization.
Usage:
@bp.get("/whatever")
@require_auth # populates g.current_user
def whatever():
return jsonify(...)
@bp.post("/admin/users")
@require_auth
@require_perm("user.create") # checks the user's effective perms
def create_user():
...
`g.current_user` is a small `AuthenticatedUser` snapshot — no live ORM session.
"""
from __future__ import annotations
import logging
import uuid
from dataclasses import dataclass, field
from functools import wraps
from typing import Callable
import jwt
from flask import abort, g, request
from sqlalchemy import select
from app.core.jwt_tokens import decode_token
from app.db.session import session_scope
from app.models.auth import Permission, User
from app.services.bootstrap import ADMIN_GROUP_NAME
log = logging.getLogger("metamorph.auth")
@dataclass(frozen=True)
class AuthenticatedUser:
id: uuid.UUID
email: str
locale: str
display_name: str | None
is_admin: bool
permissions: frozenset[str] = field(default_factory=frozenset)
group_names: frozenset[str] = field(default_factory=frozenset)
def _load_authenticated_user(user_id: uuid.UUID) -> AuthenticatedUser | None:
with session_scope() as s:
user = s.get(User, user_id)
if user is None or user.deleted_at is not None or not user.is_active:
return None
group_names: set[str] = set()
permissions: set[str] = set()
for grp in user.groups:
if grp.deleted_at is not None:
continue
group_names.add(grp.name)
for perm in grp.permissions:
permissions.add(perm.code)
return AuthenticatedUser(
id=user.id,
email=user.email,
locale=user.locale,
display_name=user.display_name,
is_admin=ADMIN_GROUP_NAME in group_names,
permissions=frozenset(permissions),
group_names=frozenset(group_names),
)
def _extract_bearer() -> str | None:
raw = request.headers.get("Authorization", "")
if not raw.lower().startswith("bearer "):
return None
return raw[7:].strip() or None
def require_auth(fn: Callable):
@wraps(fn)
def wrapper(*args, **kwargs):
token = _extract_bearer()
if token is None:
abort(401, description="missing bearer token")
try:
claims = decode_token(token, expected_type="access")
except jwt.ExpiredSignatureError:
abort(401, description="access token expired")
except jwt.PyJWTError:
abort(401, description="invalid access token")
try:
user_id = uuid.UUID(claims.sub)
except ValueError:
abort(401, description="malformed subject")
snapshot = _load_authenticated_user(user_id)
if snapshot is None:
abort(401, description="user no longer active")
g.current_user = snapshot
return fn(*args, **kwargs)
return wrapper
def require_perm(*codes: str):
"""Require any one of the listed permission codes.
Members of the system `admin` group bypass the check.
"""
def decorator(fn: Callable):
@wraps(fn)
def wrapper(*args, **kwargs):
user: AuthenticatedUser | None = getattr(g, "current_user", None)
if user is None:
abort(401, description="not authenticated")
if user.is_admin:
return fn(*args, **kwargs)
if not any(code in user.permissions for code in codes):
log.info(
"metamorph.auth.permission_denied",
extra={
"user_id": str(user.id),
"required": list(codes),
"had": sorted(user.permissions),
},
)
abort(403, description="insufficient permissions")
return fn(*args, **kwargs)
return wrapper
return decorator
def fetch_all_permissions() -> list[str]:
"""Utility for debugging / admin UI: list every known permission code."""
with session_scope() as s:
return list(s.scalars(select(Permission.code).order_by(Permission.code)).all())

147
backend/app/core/install_token.py Normal file
View File

@@ -0,0 +1,147 @@
"""First-admin install token.
When the `users` table is empty at boot, we mint a one-shot opaque token,
store its SHA-256 in `settings(key='install_token_hash')`, and log the raw
token to stdout. The operator copies it from the logs and posts it to
`/api/v1/setup` with the desired admin credentials.
Idempotency: as long as the token row exists and no admin has consumed it,
subsequent boots reuse the same hash and re-emit the same token only if
explicitly invoked via `flask metamorph print-install-token`.
"""
from __future__ import annotations
import logging
from datetime import datetime, timedelta, timezone
from sqlalchemy import select
from app.core.security import generate_opaque_token, hash_opaque_token
from app.db.session import session_scope
from app.models.auth import User
from app.models.setting import Setting
INSTALL_TOKEN_KEY = "install_token"
log = logging.getLogger("metamorph.bootstrap")
# Setting JSONB shape: {"hash": "<sha256>", "issued_at": ISO, "expires_at": ISO|null, "consumed_at": ISO|null}
def _users_exist() -> bool:
with session_scope() as s:
return s.execute(select(User.id).limit(1)).first() is not None
def _read_setting() -> Setting | None:
with session_scope() as s:
return s.get(Setting, INSTALL_TOKEN_KEY)
def _write_setting(payload: dict) -> None:
with session_scope() as s:
existing = s.get(Setting, INSTALL_TOKEN_KEY)
if existing is None:
s.add(
Setting(
key=INSTALL_TOKEN_KEY,
value=payload,
description="One-shot bootstrap token for the first admin (M2).",
)
)
else:
existing.value = payload
def ensure_install_token(*, force: bool = False) -> str | None:
"""Mint a token if no users exist and no live token is on file.
Returns the raw token if newly minted (caller is responsible for logging it),
or None if the bootstrap is already consumed / not applicable.
"""
if _users_exist() and not force:
return None
setting = _read_setting()
if setting is not None and not force:
value = setting.value or {}
if value.get("consumed_at"):
return None # consumed, do not mint again
# A pending token exists; we don't know its raw value any more.
# Caller must `force=True` to mint a new one (CLI command will do that).
return None
token = generate_opaque_token()
_write_setting(
{
"hash": hash_opaque_token(token),
"issued_at": datetime.now(tz=timezone.utc).isoformat(),
"expires_at": None, # never expires until consumed
"consumed_at": None,
}
)
return token
def regenerate_install_token() -> str:
"""CLI helper: always mint and persist a fresh token (overwrites any pending one)."""
return ensure_install_token(force=True) or _force_mint()
def _force_mint() -> str:
token = generate_opaque_token()
_write_setting(
{
"hash": hash_opaque_token(token),
"issued_at": datetime.now(tz=timezone.utc).isoformat(),
"expires_at": None,
"consumed_at": None,
}
)
return token
def verify_install_token(token: str) -> bool:
"""Constant-time comparison against the stored hash."""
setting = _read_setting()
if setting is None or not setting.value:
return False
payload = setting.value
if payload.get("consumed_at"):
return False
expected = payload.get("hash")
if not expected:
return False
import hmac
return hmac.compare_digest(hash_opaque_token(token), expected)
def mark_install_token_consumed() -> None:
setting = _read_setting()
if setting is None:
return
payload = dict(setting.value or {})
payload["consumed_at"] = datetime.now(tz=timezone.utc).isoformat()
_write_setting(payload)
def log_install_token_banner(raw_token: str) -> None:
"""Pretty banner so the token is unmissable in container logs."""
sep = "=" * 72
log.warning(
"metamorph.install_token.minted",
extra={
"banner": sep,
"message_template": (
"BOOTSTRAP — copy the token below and POST it to /api/v1/setup "
"with your desired admin email + password. Save it: it is logged once."
),
"install_token": raw_token,
},
)
# Also dump a plain banner so the token is grep-friendly even if the JSON
# consumer hides `extra` fields.
print(sep, flush=True) # noqa: T201
print(f"INSTALL TOKEN: {raw_token}", flush=True) # noqa: T201
print(sep, flush=True) # noqa: T201

97
backend/app/core/jwt_tokens.py Normal file
View File

@@ -0,0 +1,97 @@
"""JWT encoding / decoding.
Two token types:
- `access` — short-lived (1 h), in `Authorization: Bearer ...` headers, kept
client-side **in memory** only (cf. spec §M2).
- `refresh` — long-lived (30 d), in an HTTPOnly Secure SameSite=Strict cookie
scoped to `/api/v1/auth/`. Rotated on every successful refresh,
old `jti` revoked.
We sign HS256 with `settings.JWT_SECRET`. The `jti` claim links each token to
its DB row in `refresh_tokens` for revocation; access tokens are stateless.
"""
from __future__ import annotations
import secrets
import uuid
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Literal
import jwt
from app.core.config import settings
ACCESS_TOKEN_TTL = timedelta(hours=1)
REFRESH_TOKEN_TTL = timedelta(days=30)
ALGORITHM = "HS256"
ISSUER = "metamorph"
TokenType = Literal["access", "refresh"]
@dataclass(frozen=True)
class TokenClaims:
sub: str # user id (UUID as string)
type: TokenType
jti: str
iat: datetime
exp: datetime
def _now() -> datetime:
return datetime.now(tz=timezone.utc)
def generate_jti() -> str:
"""Compact, URL-safe random identifier (≈22 chars)."""
return secrets.token_urlsafe(16)
def encode_token(
user_id: uuid.UUID | str,
token_type: TokenType,
*,
jti: str | None = None,
) -> tuple[str, TokenClaims]:
"""Return `(jwt_string, claims)`. `jti` is generated if not provided."""
now = _now()
ttl = ACCESS_TOKEN_TTL if token_type == "access" else REFRESH_TOKEN_TTL
claims = TokenClaims(
sub=str(user_id),
type=token_type,
jti=jti or generate_jti(),
iat=now,
exp=now + ttl,
)
payload = {
"iss": ISSUER,
"sub": claims.sub,
"type": claims.type,
"jti": claims.jti,
"iat": int(claims.iat.timestamp()),
"exp": int(claims.exp.timestamp()),
}
return jwt.encode(payload, settings.JWT_SECRET, algorithm=ALGORITHM), claims
def decode_token(token: str, *, expected_type: TokenType) -> TokenClaims:
"""Decode and validate a JWT. Raises `jwt.PyJWTError` on any failure."""
payload = jwt.decode(
token,
settings.JWT_SECRET,
algorithms=[ALGORITHM],
issuer=ISSUER,
options={"require": ["sub", "type", "jti", "iat", "exp"]},
)
if payload["type"] != expected_type:
raise jwt.InvalidTokenError(f"expected {expected_type} token, got {payload['type']}")
return TokenClaims(
sub=payload["sub"],
type=payload["type"],
jti=payload["jti"],
iat=datetime.fromtimestamp(payload["iat"], tz=timezone.utc),
exp=datetime.fromtimestamp(payload["exp"], tz=timezone.utc),
)

29
backend/app/core/rate_limit.py Normal file
View File

@@ -0,0 +1,29 @@
"""Shared flask-limiter instance.
Anchored on remote address. In-memory backend for v1 (single-process gunicorn
worker pool can drift; that's acceptable at this scale). M14 will switch to
Redis if it becomes a real concern.
The limiter is enforced in `APP_ENV in ("prod", "staging")` — dev and test
deployments share an in-memory backend that's noisy across hot-reloads and
would gate the Playwright e2e suite at 10 req/min/IP. The spec NF-security
requirement is explicitly a *production* one (cf. tasks/spec.md §6
NF-security); a staging deployment is exposed to humans so the same limits
apply there.
"""
from __future__ import annotations
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
from app.core.config import settings
limiter = Limiter(
key_func=get_remote_address,
default_limits=[],
storage_uri="memory://",
headers_enabled=True,
strategy="fixed-window",
enabled=settings.APP_ENV in ("prod", "staging"),
)

62
backend/app/core/security.py Normal file
View File

@@ -0,0 +1,62 @@
"""Password hashing and constant-time secret hashing."""
from __future__ import annotations
import hashlib
import hmac
import secrets
from argon2 import PasswordHasher
from argon2.exceptions import VerifyMismatchError
# Argon2id with moderate cost. `time_cost=2`, `memory_cost=64MiB`, `parallelism=2`
# is well above OWASP minimums while staying snappy on a Debian small VM.
_hasher = PasswordHasher(
time_cost=2,
memory_cost=64 * 1024,
parallelism=2,
hash_len=32,
salt_len=16,
)
def hash_password(plaintext: str) -> str:
return _hasher.hash(plaintext)
def verify_password(stored_hash: str, plaintext: str) -> bool:
"""Constant-time verification. Returns False on mismatch, never raises."""
try:
return _hasher.verify(stored_hash, plaintext)
except VerifyMismatchError:
return False
except Exception: # corrupted hash or unsupported parameters
return False
def needs_rehash(stored_hash: str) -> bool:
"""True when Argon2 parameters have evolved since the hash was created."""
try:
return _hasher.check_needs_rehash(stored_hash)
except Exception:
return True
# === Opaque-token helpers (refresh tokens, invitation tokens) ===
#
# We never store the raw token in DB — only its SHA-256. Comparison uses
# `hmac.compare_digest` to dodge timing attacks. Tokens are URL-safe base64.
TOKEN_BYTES = 48 # 384 bits of entropy → 64 chars b64url
def generate_opaque_token() -> str:
return secrets.token_urlsafe(TOKEN_BYTES)
def hash_opaque_token(token: str) -> str:
return hashlib.sha256(token.encode("utf-8")).hexdigest()
def verify_opaque_token(token: str, stored_hash: str) -> bool:
return hmac.compare_digest(hash_opaque_token(token), stored_hash)

0
backend/app/services/__init__.py Normal file
View File

224
backend/app/services/auth.py Normal file
View File

@@ -0,0 +1,224 @@
"""Auth domain logic: login, refresh rotation, logout, change_password.
Returns lightweight DTOs (dicts) — the API layer is responsible for HTTP shape.
Raises plain `ValueError` / `LookupError` / `PermissionError` and lets the API
layer translate them into HTTP statuses.
"""
from __future__ import annotations
import uuid
from dataclasses import dataclass
from datetime import datetime, timezone
from sqlalchemy import select
from app.core.jwt_tokens import (
REFRESH_TOKEN_TTL,
decode_token,
encode_token,
generate_jti,
)
from app.core.security import (
hash_opaque_token,
hash_password,
needs_rehash,
verify_opaque_token,
verify_password,
)
from app.db.session import session_scope
from app.models.auth import RefreshToken, User
class AuthError(Exception):
"""Base for auth-flow exceptions; HTTP layer maps to 401/403."""
class InvalidCredentials(AuthError):
pass
class TokenRevoked(AuthError):
pass
@dataclass
class TokenPair:
access_token: str
refresh_token: str
refresh_expires_at: datetime
user_id: uuid.UUID
def _now() -> datetime:
return datetime.now(tz=timezone.utc)
# === Login ===================================================================
def login(email: str, password: str) -> TokenPair:
email_norm = email.strip().lower()
with session_scope() as s:
user = s.scalar(
select(User).where(
User.email == email_norm,
User.deleted_at.is_(None),
User.is_active.is_(True),
)
)
if user is None or not verify_password(user.password_hash, password):
# Same error for "no such user" and "wrong password" — no account enumeration.
raise InvalidCredentials("invalid credentials")
if needs_rehash(user.password_hash):
user.password_hash = hash_password(password)
return _issue_token_pair(s, user.id)
# === Refresh rotation ========================================================
def refresh(raw_refresh_token: str) -> TokenPair:
"""Validate the refresh token, revoke the old one, mint a new pair.
Detects token reuse: if a refresh token that has already been rotated is
presented again, we revoke the entire chain (treat as compromise).
"""
try:
claims = decode_token(raw_refresh_token, expected_type="refresh")
except Exception as e:
raise InvalidCredentials("invalid refresh token") from e
token_hash = hash_opaque_token(raw_refresh_token)
with session_scope() as s:
rt = s.scalar(
select(RefreshToken).where(
RefreshToken.jti == claims.jti,
RefreshToken.token_hash == token_hash,
)
)
if rt is None:
raise InvalidCredentials("refresh token not recognised")
if rt.revoked_at is not None:
# Reuse of a revoked token → likely compromise. Cascade-revoke chain.
_revoke_chain(s, rt)
raise TokenRevoked("refresh token has been revoked")
if rt.expires_at <= _now():
raise InvalidCredentials("refresh token expired")
# Rotate: mark old as revoked + replaced_by, mint new.
new_pair = _issue_token_pair(s, rt.user_id)
new_jti = decode_token(new_pair.refresh_token, expected_type="refresh").jti
new_rt = s.scalar(select(RefreshToken).where(RefreshToken.jti == new_jti))
rt.revoked_at = _now()
rt.replaced_by_id = new_rt.id if new_rt else None
return new_pair
# === Logout ==================================================================
def logout(raw_refresh_token: str) -> None:
"""Revoke the refresh token. Idempotent — silently no-ops on bad tokens."""
try:
claims = decode_token(raw_refresh_token, expected_type="refresh")
except Exception:
return
with session_scope() as s:
rt = s.scalar(select(RefreshToken).where(RefreshToken.jti == claims.jti))
if rt is not None and rt.revoked_at is None:
rt.revoked_at = _now()
def logout_all_for_user(user_id: uuid.UUID) -> int:
"""Revoke every active refresh token for a user. Returns count revoked."""
now = _now()
with session_scope() as s:
active = s.scalars(
select(RefreshToken).where(
RefreshToken.user_id == user_id,
RefreshToken.revoked_at.is_(None),
)
).all()
for rt in active:
rt.revoked_at = now
return len(active)
# === Password change ========================================================
def change_password(user_id: uuid.UUID, current: str, new: str) -> None:
if len(new) < 8:
raise ValueError("new password must be at least 8 characters")
with session_scope() as s:
user = s.get(User, user_id)
if user is None or user.deleted_at is not None or not user.is_active:
raise LookupError("user not found")
if not verify_password(user.password_hash, current):
raise InvalidCredentials("current password is incorrect")
user.password_hash = hash_password(new)
# Force re-login on every other device.
logout_all_for_user(user_id)
# === Helpers =================================================================
def _issue_token_pair(s, user_id: uuid.UUID) -> TokenPair:
"""Issue a fresh access + refresh pair. The refresh row is persisted."""
access_jti = generate_jti()
refresh_jti = generate_jti()
access_token, _ = encode_token(user_id, "access", jti=access_jti)
refresh_token, refresh_claims = encode_token(user_id, "refresh", jti=refresh_jti)
s.add(
RefreshToken(
user_id=user_id,
jti=refresh_jti,
token_hash=hash_opaque_token(refresh_token),
issued_at=refresh_claims.iat,
expires_at=refresh_claims.exp,
)
)
s.flush() # ensure the row gets an id before we return
return TokenPair(
access_token=access_token,
refresh_token=refresh_token,
refresh_expires_at=refresh_claims.exp,
user_id=user_id,
)
def _revoke_chain(s, rt: RefreshToken) -> None:
"""When reuse is detected, revoke this token and its replacement chain."""
seen: set[uuid.UUID] = set()
cur: RefreshToken | None = rt
while cur is not None and cur.id not in seen:
seen.add(cur.id)
if cur.revoked_at is None:
cur.revoked_at = _now()
if cur.replaced_by_id:
cur = s.get(RefreshToken, cur.replaced_by_id)
else:
cur = None
__all__ = [
"AuthError",
"InvalidCredentials",
"TokenRevoked",
"TokenPair",
"REFRESH_TOKEN_TTL",
"login",
"refresh",
"logout",
"logout_all_for_user",
"change_password",
]

98
backend/app/services/bootstrap.py Normal file
View File

@@ -0,0 +1,98 @@
"""Initial bootstrap : seed `admin` / `redteam` / `blueteam` system groups + first admin.
The detailed permission seeding lives in M3 (`mitre.sync` etc.); for M2 we only
need an `admin` group that effectively grants full access. We model that as an
absent permission set + a special `is_system` flag on the group, plus the
`@require_perm` decorator that bypasses checks for any user belonging to a
system `admin` group. M3 will fill in the atomic permissions.
"""
from __future__ import annotations
import uuid
from dataclasses import dataclass
from sqlalchemy import select
from app.core.install_token import (
mark_install_token_consumed,
verify_install_token,
)
from app.core.security import hash_password
from app.db.session import session_scope
from app.models.auth import Group, User, UserGroup
ADMIN_GROUP_NAME = "admin"
REDTEAM_GROUP_NAME = "redteam"
BLUETEAM_GROUP_NAME = "blueteam"
@dataclass
class BootstrapResult:
user_id: uuid.UUID
admin_group_id: uuid.UUID
class BootstrapError(Exception):
pass
def ensure_system_groups() -> dict[str, uuid.UUID]:
"""Create the three system groups if missing. Idempotent."""
out: dict[str, uuid.UUID] = {}
with session_scope() as s:
for name, desc in (
(ADMIN_GROUP_NAME, "Platform administrators — full access."),
(REDTEAM_GROUP_NAME, "Red team operators."),
(BLUETEAM_GROUP_NAME, "Blue team operators."),
):
grp = s.scalar(select(Group).where(Group.name == name, Group.is_system.is_(True)))
if grp is None:
grp = Group(name=name, description=desc, is_system=True)
s.add(grp)
s.flush()
out[name] = grp.id
return out
def bootstrap_admin(
*, install_token: str, email: str, password: str, display_name: str | None = None
) -> BootstrapResult:
"""Consume the install token, create the first admin user, attach to admin group."""
if not verify_install_token(install_token):
raise BootstrapError("invalid or already-consumed install token")
if len(password) < 8:
raise ValueError("password must be at least 8 characters")
email_norm = email.strip().lower()
# Re-check users count under transaction to avoid races.
with session_scope() as s:
if s.scalar(select(User.id).limit(1)) is not None:
raise BootstrapError("setup already done — at least one user exists")
groups = ensure_system_groups()
with session_scope() as s:
user = User(
email=email_norm,
display_name=(display_name or "").strip() or None,
password_hash=hash_password(password),
)
s.add(user)
s.flush()
s.add(UserGroup(user_id=user.id, group_id=groups[ADMIN_GROUP_NAME]))
admin_id = groups[ADMIN_GROUP_NAME]
user_id = user.id
mark_install_token_consumed()
# Re-seed the permission catalogue + system-group bindings. This is called
# at boot too, but on a fresh DB after `/diag/reset` the groups were just
# recreated above and have no permissions yet — seeding here keeps the
# bootstrap path self-contained.
from app.services.permissions_seed import seed_all # noqa: PLC0415 — avoid import cycle
seed_all()
return BootstrapResult(user_id=user_id, admin_group_id=admin_id)

188
backend/app/services/invitations.py Normal file
View File

@@ -0,0 +1,188 @@
"""Invitation flow: admin issues a one-shot URL token, invitee accepts.
The raw token is shown to the admin once (returned by `create_invitation`)
and never persisted — only its SHA-256 lives in the DB. Pre-assigned groups
are attached at creation and applied at acceptance.
"""
from __future__ import annotations
import uuid
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Iterable
from sqlalchemy import select
from app.core.security import (
generate_opaque_token,
hash_opaque_token,
hash_password,
)
from app.db.session import session_scope
from app.models.auth import Group, Invitation, InvitationGroup, User, UserGroup
INVITATION_TTL = timedelta(days=7)
class InvitationError(Exception):
pass
class InvitationExpired(InvitationError):
pass
class InvitationConsumed(InvitationError):
pass
class InvitationRevoked(InvitationError):
pass
@dataclass
class InvitationCreated:
invitation_id: uuid.UUID
raw_token: str
expires_at: datetime
@dataclass
class InvitationPreview:
email_hint: str | None
expires_at: datetime
groups: list[str]
is_valid: bool
reason: str | None
def _now() -> datetime:
return datetime.now(tz=timezone.utc)
def create_invitation(
*,
created_by_user_id: uuid.UUID,
email_hint: str | None,
group_ids: Iterable[uuid.UUID] = (),
ttl: timedelta = INVITATION_TTL,
) -> InvitationCreated:
raw = generate_opaque_token()
expires_at = _now() + ttl
with session_scope() as s:
inv = Invitation(
token_hash=hash_opaque_token(raw),
email_hint=email_hint.strip().lower() if email_hint else None,
created_by_user_id=created_by_user_id,
expires_at=expires_at,
)
s.add(inv)
s.flush()
for gid in group_ids:
s.add(InvitationGroup(invitation_id=inv.id, group_id=gid))
return InvitationCreated(
invitation_id=inv.id,
raw_token=raw,
expires_at=expires_at,
)
def _load_by_token(s, raw_token: str) -> Invitation | None:
return s.scalar(
select(Invitation).where(Invitation.token_hash == hash_opaque_token(raw_token))
)
def preview(raw_token: str) -> InvitationPreview:
with session_scope() as s:
inv = _load_by_token(s, raw_token)
if inv is None:
return InvitationPreview(None, _now(), [], False, "not_found")
groups = [g.name for g in inv.pre_assigned_groups]
if inv.revoked_at is not None:
return InvitationPreview(inv.email_hint, inv.expires_at, groups, False, "revoked")
if inv.consumed_at is not None:
return InvitationPreview(inv.email_hint, inv.expires_at, groups, False, "consumed")
if inv.expires_at <= _now():
return InvitationPreview(inv.email_hint, inv.expires_at, groups, False, "expired")
return InvitationPreview(inv.email_hint, inv.expires_at, groups, True, None)
def accept(raw_token: str, *, email: str, password: str, display_name: str | None) -> uuid.UUID:
"""Create the user, attach pre-assigned groups, mark invitation consumed."""
if len(password) < 8:
raise ValueError("password must be at least 8 characters")
email_norm = email.strip().lower()
with session_scope() as s:
inv = _load_by_token(s, raw_token)
if inv is None:
raise InvitationError("invitation not found")
if inv.revoked_at is not None:
raise InvitationRevoked("invitation revoked")
if inv.consumed_at is not None:
raise InvitationConsumed("invitation already consumed")
if inv.expires_at <= _now():
raise InvitationExpired("invitation expired")
# Email must not be already in use among active users.
existing = s.scalar(
select(User).where(User.email == email_norm, User.deleted_at.is_(None))
)
if existing is not None:
raise ValueError("email already in use")
user = User(
email=email_norm,
display_name=(display_name or "").strip() or None,
password_hash=hash_password(password),
)
s.add(user)
s.flush()
for grp in inv.pre_assigned_groups:
s.add(UserGroup(user_id=user.id, group_id=grp.id))
inv.consumed_at = _now()
inv.consumed_by_user_id = user.id
return user.id
def revoke(invitation_id: uuid.UUID) -> bool:
with session_scope() as s:
inv = s.get(Invitation, invitation_id)
if inv is None:
return False
if inv.revoked_at is not None or inv.consumed_at is not None:
return False
inv.revoked_at = _now()
return True
def list_active(*, limit: int = 100) -> list[Invitation]:
with session_scope() as s:
rows = s.scalars(
select(Invitation)
.where(
Invitation.consumed_at.is_(None),
Invitation.revoked_at.is_(None),
Invitation.expires_at > _now(),
)
.order_by(Invitation.created_at.desc())
.limit(limit)
).all()
# detach so caller can read after session closes
for r in rows:
s.expunge(r)
for g in r.pre_assigned_groups:
s.expunge(g)
return list(rows)
def find_group_id_by_name(name: str) -> uuid.UUID | None:
with session_scope() as s:
gid = s.scalar(
select(Group.id).where(Group.name == name, Group.deleted_at.is_(None))
)
return gid

299
backend/tests/test_auth_flow.py Normal file
View File

@@ -0,0 +1,299 @@
"""End-to-end auth flow integration test (live DB).
Hits the Flask test client to exercise:
- /setup with the install token
- /auth/login + /auth/me
- /auth/refresh (rotation)
- /auth/logout (revocation, idempotency)
- /auth/change-password (forces logout-all)
- /invitations create + preview + accept
- RBAC: non-admin gets 403 on admin endpoint
The DB schema is left in place between tests; we use unique emails to avoid
collisions across runs. The install token is force-minted at the start.
"""
from __future__ import annotations
import json
import secrets
import uuid
import pytest
from sqlalchemy import text
from app.core.install_token import regenerate_install_token
from app.db.session import get_engine
from app.main import create_app
def _truncate_users(engine):
"""Wipe data so /setup has work to do. CASCADE handles dependent rows."""
with engine.begin() as conn:
conn.execute(
text(
"TRUNCATE users, refresh_tokens, invitations, invitation_groups, "
"user_groups, settings, groups RESTART IDENTITY CASCADE"
)
)
@pytest.fixture(scope="module")
def app(db_engine_or_skip):
_truncate_users(db_engine_or_skip)
flask_app = create_app()
flask_app.config.update(TESTING=True)
return flask_app
@pytest.fixture()
def client(app):
return app.test_client()
@pytest.fixture(scope="module")
def install_token(db_engine_or_skip):
return regenerate_install_token()
def _unique_email(prefix: str) -> str:
return f"{prefix}-{secrets.token_hex(4)}@metamorph.local"
# -- /setup -------------------------------------------------------------------
def test_setup_status_starts_uncompleted(client):
r = client.get("/api/v1/setup")
assert r.status_code == 200
assert r.get_json()["completed"] is False
def test_setup_creates_first_admin(client, install_token):
email = _unique_email("admin")
r = client.post(
"/api/v1/setup",
json={
"install_token": install_token,
"email": email,
"password": "AdminPass1234!",
"display_name": "Init Admin",
},
)
assert r.status_code == 201, r.get_data(as_text=True)
body = r.get_json()
assert "user_id" in body
pytest.shared_admin = {"email": email, "password": "AdminPass1234!", "id": body["user_id"]} # type: ignore[attr-defined]
def test_setup_status_now_completed(client):
assert client.get("/api/v1/setup").get_json()["completed"] is True
def test_setup_replay_is_blocked(client, install_token):
# Token already consumed — a second call must be refused.
r = client.post(
"/api/v1/setup",
json={
"install_token": install_token,
"email": _unique_email("replay"),
"password": "AdminPass1234!",
},
)
assert r.status_code == 409
# -- /auth/login + /auth/me ---------------------------------------------------
@pytest.fixture(scope="module")
def admin_credentials():
return getattr(pytest, "shared_admin") # populated by test_setup_creates_first_admin
def _login(client, email: str, password: str) -> tuple[str, dict]:
r = client.post("/api/v1/auth/login", json={"email": email, "password": password})
assert r.status_code == 200, r.get_data(as_text=True)
body = r.get_json()
return body["access_token"], dict(r.headers)
def test_login_and_me(client, admin_credentials):
access, _ = _login(client, admin_credentials["email"], admin_credentials["password"])
r = client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {access}"})
assert r.status_code == 200
me = r.get_json()
assert me["email"] == admin_credentials["email"]
assert me["is_admin"] is True
assert "admin" in me["groups"]
def test_login_with_wrong_password_returns_401(client, admin_credentials):
r = client.post(
"/api/v1/auth/login",
json={"email": admin_credentials["email"], "password": "wrong"},
)
assert r.status_code == 401
assert r.get_json()["error"] == "invalid_credentials"
def test_me_without_token_returns_401(client):
r = client.get("/api/v1/auth/me")
assert r.status_code == 401
# -- /auth/refresh rotation ---------------------------------------------------
def test_refresh_rotates_and_old_token_is_revoked(client, admin_credentials):
# Login fresh to get a refresh cookie on the test client.
client.post(
"/api/v1/auth/login",
json={"email": admin_credentials["email"], "password": admin_credentials["password"]},
)
# First refresh — should succeed.
r1 = client.post("/api/v1/auth/refresh")
assert r1.status_code == 200, r1.get_data(as_text=True)
new_access1 = r1.get_json()["access_token"]
assert new_access1
# Second refresh — uses the rotated cookie automatically (test client persists cookies).
r2 = client.post("/api/v1/auth/refresh")
assert r2.status_code == 200
assert r2.get_json()["access_token"] != new_access1
def test_refresh_with_no_cookie_returns_401(client):
fresh = client.application.test_client() # blank cookie jar
r = fresh.post("/api/v1/auth/refresh")
assert r.status_code == 401
# -- /auth/logout -------------------------------------------------------------
def test_logout_clears_cookie_and_is_idempotent(client, admin_credentials):
client.post(
"/api/v1/auth/login",
json={"email": admin_credentials["email"], "password": admin_credentials["password"]},
)
r1 = client.post("/api/v1/auth/logout")
assert r1.status_code == 200
# Second logout — no token, still 200.
r2 = client.post("/api/v1/auth/logout")
assert r2.status_code == 200
# -- /invitations -------------------------------------------------------------
def test_admin_creates_invitation_and_invitee_accepts(client, admin_credentials):
access, _ = _login(client, admin_credentials["email"], admin_credentials["password"])
inv_email = _unique_email("alice")
create = client.post(
"/api/v1/invitations",
headers={"Authorization": f"Bearer {access}"},
json={"email_hint": inv_email},
)
assert create.status_code == 201, create.get_data(as_text=True)
token = create.get_json()["token"]
preview = client.get(f"/api/v1/invitations/preview/{token}")
assert preview.status_code == 200
assert preview.get_json()["is_valid"] is True
accept = client.post(
f"/api/v1/invitations/accept/{token}",
json={"email": inv_email, "password": "AlicePass1234!", "display_name": "Alice"},
)
assert accept.status_code == 201, accept.get_data(as_text=True)
# Alice can now log in.
a_access, _ = _login(client, inv_email, "AlicePass1234!")
me = client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {a_access}"}).get_json()
assert me["is_admin"] is False
assert me["email"] == inv_email
def test_unauthenticated_cannot_create_invitation(client):
r = client.post("/api/v1/invitations", json={})
assert r.status_code == 401
def test_non_admin_cannot_create_invitation(client, admin_credentials):
# Create a non-admin user via invitation, then try as them.
access, _ = _login(client, admin_credentials["email"], admin_credentials["password"])
inv_email = _unique_email("bob")
create = client.post(
"/api/v1/invitations",
headers={"Authorization": f"Bearer {access}"},
json={"email_hint": inv_email},
)
token = create.get_json()["token"]
client.post(
f"/api/v1/invitations/accept/{token}",
json={"email": inv_email, "password": "BobPass1234!"},
)
bob_access, _ = _login(client, inv_email, "BobPass1234!")
r = client.post(
"/api/v1/invitations",
headers={"Authorization": f"Bearer {bob_access}"},
json={},
)
assert r.status_code == 403
def test_used_invitation_cannot_be_accepted_twice(client, admin_credentials):
access, _ = _login(client, admin_credentials["email"], admin_credentials["password"])
inv_email = _unique_email("carol")
token = client.post(
"/api/v1/invitations",
headers={"Authorization": f"Bearer {access}"},
json={"email_hint": inv_email},
).get_json()["token"]
first = client.post(
f"/api/v1/invitations/accept/{token}",
json={"email": inv_email, "password": "CarolPass1234!"},
)
assert first.status_code == 201
second = client.post(
f"/api/v1/invitations/accept/{token}",
json={"email": _unique_email("carol2"), "password": "OtherPass1234!"},
)
assert second.status_code == 410
assert second.get_json()["error"] == "invitation_consumed"
# -- change-password forces logout-all ----------------------------------------
def test_change_password_revokes_all_refresh_tokens(client, admin_credentials):
access, _ = _login(client, admin_credentials["email"], admin_credentials["password"])
# Trigger a couple of refreshes so we have multiple chains in DB.
client.post("/api/v1/auth/refresh")
client.post("/api/v1/auth/refresh")
# Change password.
new_pw = "AdminPass5678!"
r = client.post(
"/api/v1/auth/change-password",
headers={"Authorization": f"Bearer {access}"},
json={
"current_password": admin_credentials["password"],
"new_password": new_pw,
},
)
assert r.status_code == 200
admin_credentials["password"] = new_pw
# Existing refresh cookie must now be rejected.
r2 = client.post("/api/v1/auth/refresh")
assert r2.status_code == 401
# New login still works.
_login(client, admin_credentials["email"], admin_credentials["password"])