Skip to content

Core Services

Activation

activation

Activation-link helpers - ported from MasterplanOptimiserV2 Server. Tokens stored as SHA-256 hashes so a DB dump never reveals a usable token.

ActivationDeliveryInProgressError

Bases: RuntimeError

Raised when manual link creation would disrupt an SMTP hand-off.

Source code in backend/app/core/activation.py 
25
26
class ActivationDeliveryInProgressError(RuntimeError):
    """Raised when manual link creation would disrupt an SMTP hand-off."""

hash_token 

hash_token(token: str) -> str

Return the hex-digest SHA-256 of token.

Source code in backend/app/core/activation.py 
29
30
31
def hash_token(token: str) -> str:
    """Return the hex-digest SHA-256 of *token*."""
    return hashlib.sha256(token.encode()).hexdigest()

resolve_activation_purpose 

resolve_activation_purpose(*, is_activated: bool, requested: ManagedPasskeyPurpose | None) -> ActivationPurpose

Resolve a safe link purpose from account state and an optional request.

Pending accounts always use initial setup. Active accounts retain the historical reset default when an older client omits the purpose.

Source code in backend/app/core/activation.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def resolve_activation_purpose(
    *,
    is_activated: bool,
    requested: ManagedPasskeyPurpose | None,
) -> ActivationPurpose:
    """Resolve a safe link purpose from account state and an optional request.

    Pending accounts always use initial setup. Active accounts retain the
    historical reset default when an older client omits the purpose.
    """

    if not is_activated:
        if requested is not None:
            raise ValueError(
                "Credential-management links require an activated account"
            )
        return INITIAL_SETUP
    return requested or CREDENTIAL_RESET

create_activation_link(user_id: int, created_by_id: int, db: Session, purpose: str = 'initial_setup', expiry_hours: int | None = None, delivery_pending: bool = False, permit_email_delivery_start: bool = False) -> Tuple[str, ActivationLink]

Create an activation link for a user.

Automatically invalidates any previous active link for the same user. Email links may be held pending until SMTP acceptance is confirmed. Only the email delivery workflow may set permit_email_delivery_start so a manual link cannot invalidate a token while its email is being handed off. Returns (raw_token, link_row).

Source code in backend/app/core/activation.py 
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def create_activation_link(
    user_id: int,
    created_by_id: int,
    db: Session,
    purpose: str = "initial_setup",
    expiry_hours: int | None = None,
    delivery_pending: bool = False,
    permit_email_delivery_start: bool = False,
) -> Tuple[str, ActivationLink]:
    """Create an activation link for a user.

    Automatically invalidates any previous active link for the same user.
    Email links may be held pending until SMTP acceptance is confirmed. Only
    the email delivery workflow may set ``permit_email_delivery_start`` so a
    manual link cannot invalidate a token while its email is being handed off.
    Returns (raw_token, link_row).
    """
    if expiry_hours is None:
        expiry_hours = runtime_settings.get_int("activation_link_expiry_hours", db)
    if purpose not in {INITIAL_SETUP, ADDITIONAL_PASSKEY, CREDENTIAL_RESET}:
        raise ValueError("Unsupported activation purpose")
    now = datetime.now(timezone.utc)

    # Serialise concurrent link creation for one user.
    user = db.query(User).filter(User.id == user_id).with_for_update().first()
    if user is None:
        raise ValueError("User not found")
    if not permit_email_delivery_start and db.query(ActivationEmailDelivery).filter(
        ActivationEmailDelivery.user_id == user_id,
        ActivationEmailDelivery.status == "sending",
    ).first() is not None:
        raise ActivationDeliveryInProgressError(
            "An activation email is already being handed off for this user"
        )

    # Invalidate all previous active links for this user
    db.query(ActivationLink).filter(
        ActivationLink.user_id == user_id,
        ActivationLink.used_at.is_(None),
        ActivationLink.invalidated_at.is_(None),
    ).update({"invalidated_at": now}, synchronize_session="fetch")

    raw_token = secrets.token_urlsafe(48)
    link = ActivationLink(
        token_hash=hash_token(raw_token),
        user_id=user_id,
        purpose=purpose,
        expires_at=now + timedelta(hours=expiry_hours),
        delivery_pending=delivery_pending,
        created_by_id=created_by_id,
    )
    db.add(link)
    db.flush()

    return raw_token, link

validate_activation_token 

validate_activation_token(token: str, db: Session, *, for_update: bool = False) -> Optional[ActivationLink]

Look up a token and return the link row if it is still valid.

Source code in backend/app/core/activation.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def validate_activation_token(
    token: str,
    db: Session,
    *,
    for_update: bool = False,
) -> Optional[ActivationLink]:
    """Look up a token and return the link row if it is still valid."""
    hashed = hash_token(token)
    query = db.query(ActivationLink).filter(ActivationLink.token_hash == hashed)
    if for_update:
        query = query.with_for_update()
    link = query.first()
    if link is None:
        return None

    now = datetime.now(timezone.utc)
    expires = link.expires_at
    if expires.tzinfo is None:
        expires = expires.replace(tzinfo=timezone.utc)

    if now > expires:
        return None
    if link.used_at is not None:
        return None
    if link.invalidated_at is not None:
        return None
    if link.delivery_pending:
        return None

    return link

mark_link_used(link: ActivationLink, db: Session) -> None

Mark an activation link as used.

Source code in backend/app/core/activation.py 
143
144
145
146
147
148
def mark_link_used(link: ActivationLink, db: Session) -> None:
    """Mark an activation link as used."""
    if link.used_at is not None or link.invalidated_at is not None:
        raise ValueError("Activation link is no longer available")
    link.used_at = datetime.now(timezone.utc)
    db.commit()

Audit

audit

Audit helper - single function to record security-relevant actions.

Usage::

from app.core.audit import audit
audit(db, user=current_user, action="event.create", resource_type="event",
      resource_id=new_event.id, request=request)

The entry is added to the session but NOT committed - the caller's existing transaction will include it.

audit 

audit(db: Session, *, user: Optional[User], action: str, resource_type: Optional[str] = None, resource_id: Optional[int] = None, detail: Optional[str] = None, request: Optional[Request] = None, outcome: str = 'success') -> AuditLog

Create one schema-bound, minimised audit entry (uncommitted).

Source code in backend/app/core/audit.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
def audit(
    db: Session,
    *,
    user: Optional[User],
    action: str,
    resource_type: Optional[str] = None,
    resource_id: Optional[int] = None,
    detail: Optional[str] = None,
    request: Optional[Request] = None,
    outcome: str = "success",
) -> AuditLog:
    """Create one schema-bound, minimised audit entry (uncommitted)."""
    if action not in AUDIT_ACTIONS:
        raise ValueError(f"unsupported audit action: {action}")
    if resource_type is not None and resource_type not in AUDIT_RESOURCE_TYPES:
        raise ValueError(f"unsupported audit resource type: {resource_type}")
    if outcome not in AUDIT_OUTCOMES:
        raise ValueError(f"unsupported audit outcome: {outcome}")
    ip_hash = None
    if request is not None:
        ip_hash = _hash_ip(request.client.host if request.client else None)

    entry = AuditLog(
        user_id=user.id if user else None,
        username=None,
        actor_ref=user.evidence_subject_id if user else None,
        action=action,
        resource_type=resource_type,
        resource_id=resource_id,
        detail=_canonical_detail(detail),
        ip_hash=ip_hash,
        outcome=outcome,
    )
    db.add(entry)
    return entry

Diff

diff

Schedule diff helpers - compute per-person change summaries between publishes.

compute_per_person_diffs 

compute_per_person_diffs(old_tasks: List[PublishedTask], old_edits_map: Dict[int, TaskEdit], new_tasks: List[PublishedTask], new_edits_map: Optional[Dict[int, TaskEdit]] = None) -> Dict[int, dict]

Compare old resolved tasks against new live tasks per person.

Returns {person_id: changes_dict} for persons with actual changes. changes_dict has keys: type, summary, added, removed, modified.

Source code in backend/app/core/diff.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def compute_per_person_diffs(
    old_tasks: List[PublishedTask],
    old_edits_map: Dict[int, TaskEdit],
    new_tasks: List[PublishedTask],
    new_edits_map: Optional[Dict[int, TaskEdit]] = None,
) -> Dict[int, dict]:
    """Compare old resolved tasks against new live tasks per person.

    Returns {person_id: changes_dict} for persons with actual changes.
    changes_dict has keys: type, summary, added, removed, modified.
    """
    old_map = _build_person_task_map(old_tasks, old_edits_map)
    new_map = _build_person_task_map(new_tasks, new_edits_map or {})

    all_person_ids = set(old_map.keys()) | set(new_map.keys())
    is_initial = len(old_tasks) == 0

    diffs: Dict[int, dict] = {}

    for pid in all_person_ids:
        old_person_tasks = old_map.get(pid, {})
        new_person_tasks = new_map.get(pid, {})

        old_ids = set(old_person_tasks.keys())
        new_ids = set(new_person_tasks.keys())

        added = [_task_summary(new_person_tasks[tid]) for tid in sorted(new_ids - old_ids)]
        removed = [_task_summary(old_person_tasks[tid]) for tid in sorted(old_ids - new_ids)]

        modified = []
        for tid in sorted(old_ids & new_ids):
            old_t = old_person_tasks[tid]
            new_t = new_person_tasks[tid]
            changes = {}
            for field in _DIFF_FIELDS:
                old_val = old_t.get(field)
                new_val = new_t.get(field)
                if old_val != new_val:
                    changes[field] = {"old": old_val, "new": new_val}
            # Check attendee list change (just names for readability)
            old_names = sorted(a.get("name", "") for a in old_t.get("attendees", []))
            new_names = sorted(a.get("name", "") for a in new_t.get("attendees", []))
            if old_names != new_names:
                changes["attendees"] = {
                    "old": ", ".join(old_names),
                    "new": ", ".join(new_names),
                }
            if changes:
                modified.append({
                    "name": new_t["name"],
                    "changes": changes,
                })

        if not added and not removed and not modified:
            continue

        total = len(added) + len(removed) + len(modified)
        change_type = "initial" if is_initial else "republish"
        summary = (
            f"Your schedule has been published with {total} task{'s' if total != 1 else ''}"
            if is_initial
            else f"{total} change{'s' if total != 1 else ''} to your schedule"
        )

        diffs[pid] = {
            "type": change_type,
            "summary": summary,
            "added": added,
            "removed": removed,
            "modified": modified,
        }

    return diffs

store_schedule_changes 

store_schedule_changes(event_id: int, diffs: Dict[int, dict], db: Session) -> int

Store per-person diffs as ScheduleChange records for linked users. Returns number of records created.

Source code in backend/app/core/diff.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def store_schedule_changes(
    event_id: int,
    diffs: Dict[int, dict],
    db: Session,
) -> int:
    """Store per-person diffs as ScheduleChange records for linked users.
    Returns number of records created."""
    if not diffs:
        return 0

    from app.models.notification import ScheduleChange

    person_ids = list(diffs.keys())
    # Find users linked to these persons for this event
    linked_users = (
        db.query(User)
        .filter(
            User.event_id == event_id,
            User.linked_person_id.in_(person_ids),
        )
        .all()
    )

    count = 0
    for user in linked_users:
        person_diff = diffs.get(user.linked_person_id)
        if person_diff is None:
            continue
        db.add(ScheduleChange(
            user_id=user.id,
            event_id=event_id,
            changes_json=json.dumps(person_diff, default=str),
        ))
        count += 1

    return count

Permissions

permissions

Permission enforcement middleware for V3 Server.

Simplified from V2: no desktop/web mode distinction. - Unauthenticated paths (passkey, activation, publish) always pass through. - All other writes require a valid session cookie + CSRF token. - Admin endpoints require admin role (enforced by route dependencies). - Calendar edits require can_edit flag (enforced by route dependencies).

Returns JSONResponse (not raise HTTPException) so outer CORSMiddleware can still add CORS headers on denied requests.

enforce_permissions_middleware async 

enforce_permissions_middleware(request: Request, call_next)

Enforce CSRF on cookie-authenticated write requests.

Route-level dependencies handle role checks (require_admin, can_edit). This middleware only ensures CSRF protection on writes that use cookies.

Source code in backend/app/core/permissions.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
async def enforce_permissions_middleware(request: Request, call_next):
    """Enforce CSRF on cookie-authenticated write requests.

    Route-level dependencies handle role checks (require_admin, can_edit).
    This middleware only ensures CSRF protection on writes that use cookies.
    """
    if request.method in WRITE_METHODS:
        path = request.url.path

        # Always allow unauthenticated / non-cookie auth paths
        if path in ALWAYS_ALLOWED_WRITE_PATHS:
            return await call_next(request)
        if (
            path in {
                "/api/v1/passkey/register/begin",
                "/api/v1/passkey/register/complete",
            }
            and request.headers.get("x-activation-token")
        ):
            return await call_next(request)
        if _matches_prefix(path, ALWAYS_ALLOWED_WRITE_PREFIXES):
            return await call_next(request)

        # CSRF check for cookie-authenticated writes
        has_session_cookie = settings.SESSION_COOKIE_NAME in request.cookies
        if has_session_cookie and not _verify_csrf(request):
            return JSONResponse(
                status_code=403,
                content={"detail": "CSRF token missing or invalid"},
            )

    return await call_next(request)

Push

push

Web Push helper - send push notifications to subscribed users. Uses pywebpush with VAPID authentication.

VAPID_PRIVATE_KEY should be a base64url-encoded raw 32-byte EC private key (the same format py-vapid and many VAPID generators output).

get_application_server_key 

get_application_server_key() -> Optional[str]

Return the VAPID public key in base64url format for the Push API.

Source code in backend/app/core/push.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def get_application_server_key() -> Optional[str]:
    """Return the VAPID public key in base64url format for the Push API."""
    global _public_key_cache
    if not _vapid_configured():
        return None
    if _public_key_cache:
        return _public_key_cache
    try:
        from cryptography.hazmat.primitives.asymmetric import ec
        from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat

        # Decode raw 32-byte private scalar
        raw = base64.urlsafe_b64decode(settings.VAPID_PRIVATE_KEY + "==")
        private_key = ec.derive_private_key(
            int.from_bytes(raw, "big"),
            ec.SECP256R1(),
        )
        pub_bytes = private_key.public_key().public_bytes(
            encoding=Encoding.X962,
            format=PublicFormat.UncompressedPoint,
        )
        _public_key_cache = base64.urlsafe_b64encode(pub_bytes).rstrip(b"=").decode("ascii")
        return _public_key_cache
    except Exception as exc:
        logger.error("Failed to derive VAPID public key (%s)", type(exc).__name__)
        return None

send_push 

send_push(endpoint: str, p256dh: str, auth: str, payload: dict) -> bool | None

Send one push, returning false only for an expired subscription.

Source code in backend/app/core/push.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def send_push(endpoint: str, p256dh: str, auth: str, payload: dict) -> bool | None:
    """Send one push, returning false only for an expired subscription."""
    if not _vapid_configured():
        return None

    try:
        # pywebpush accepts raw base64url private key string directly
        webpush(
            subscription_info={
                "endpoint": endpoint,
                "keys": {
                    "p256dh": p256dh,
                    "auth": auth,
                },
            },
            data=json.dumps(payload),
            vapid_private_key=settings.VAPID_PRIVATE_KEY,
            vapid_claims={
                "sub": settings.VAPID_CLAIMS_EMAIL
                if settings.VAPID_CLAIMS_EMAIL.startswith("mailto:")
                else f"mailto:{settings.VAPID_CLAIMS_EMAIL}"
            },
        )
        return True
    except WebPushException as exc:
        # 410 Gone or 404 = subscription expired, caller should delete
        if hasattr(exc, "response") and exc.response is not None:
            if exc.response.status_code in (404, 410):
                return False
        status_code = getattr(getattr(exc, "response", None), "status_code", None)
        logger.warning(
            "Push delivery failed (%s, status=%s)",
            type(exc).__name__,
            status_code,
        )
        return None
    except Exception as exc:
        logger.warning("Push delivery failed (%s)", type(exc).__name__)
        return None

send_push_to_event 

send_push_to_event(event_id: int, title: str, body: str, url: Optional[str], db: Session, notification_type: Optional[str] = None) -> int

Send push to all subscribers of an event. Returns count of successful deliveries. Removes expired subscriptions (410/404). notification_type: "announcement" or "schedule" (used by SW to pick icon).

Source code in backend/app/core/push.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def send_push_to_event(
    event_id: int, title: str, body: str, url: Optional[str], db: Session,
    notification_type: Optional[str] = None,
) -> int:
    """Send push to all subscribers of an event. Returns count of successful deliveries.
    Removes expired subscriptions (410/404).
    notification_type: "announcement" or "schedule" (used by SW to pick icon)."""
    from app.models.notification import PushSubscription

    if not _vapid_configured():
        logger.info("VAPID is not configured; push delivery skipped")
        return 0

    subs = db.query(PushSubscription).filter(PushSubscription.event_id == event_id).all()
    if not subs:
        return 0

    payload = {"title": title, "body": body}
    if url:
        payload["url"] = url
    if notification_type:
        payload["type"] = notification_type

    sent = 0
    expired_ids = []
    for sub in subs:
        ok = send_push(sub.endpoint, sub.p256dh, sub.auth, payload)
        if ok:
            sent += 1
        elif ok is False:
            expired_ids.append(sub.id)

    # Clean up expired subscriptions
    if expired_ids:
        db.query(PushSubscription).filter(PushSubscription.id.in_(expired_ids)).delete(synchronize_session=False)
        db.commit()

    logger.info(
        "Push delivery completed for event %s: %s/%s sent",
        event_id,
        sent,
        len(subs),
    )
    return sent

Runtime Settings

runtime_settings

Runtime-configurable security settings.

Reads overrides from the server_settings DB table and falls back to the static values in config.py / hard-coded defaults. Every public getter accepts an optional db session so callers that already have one can avoid opening a second connection.

get_all 

get_all(db: Session) -> Dict[str, dict]

Return every tuneable setting with its current effective value and metadata.

Source code in backend/app/core/runtime_settings.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def get_all(db: Session) -> Dict[str, dict]:
    """Return every tuneable setting with its current effective value and metadata."""
    overrides = _get_overrides(db)
    result = {}
    for key, meta in TUNEABLE_SETTINGS.items():
        raw = overrides.get(key)
        if raw is not None:
            try:
                value = meta["type"](raw)
                if not meta["min"] <= value <= meta["max"]:
                    value = meta["default"]
            except (ValueError, TypeError):
                value = meta["default"]
        else:
            value = meta["default"]
        item = {
            "value": value,
            "default": meta["default"],
            "label": meta["label"],
            "unit": meta["unit"],
            "min": meta["min"],
            "max": meta["max"],
        }
        if key in GOVERNANCE_RUNTIME_FIELDS:
            item["governance_managed"] = True
            item["governance_field"] = GOVERNANCE_RUNTIME_FIELDS[key][0]
        result[key] = item
    return result

get_int 

get_int(key: str, db: Session) -> int

Return the effective integer value for key.

Source code in backend/app/core/runtime_settings.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def get_int(key: str, db: Session) -> int:
    """Return the effective integer value for *key*."""
    meta = TUNEABLE_SETTINGS.get(key)
    if meta is None:
        raise KeyError(f"Unknown setting: {key}")
    from app.models.server_setting import ServerSetting
    row = db.query(ServerSetting).filter(ServerSetting.key == key).first()
    if row is not None:
        try:
            v = int(row.value)
            if meta["min"] <= v <= meta["max"]:
                return v
        except (ValueError, TypeError):
            pass
    return meta["default"]

apply_governance_runtime_values 

apply_governance_runtime_values(structured: dict[str, Any], db: Session) -> tuple[dict[str, Any], list[dict[str, Any]]]

Overlay only settings the Server actually enforces onto a draft.

Source code in backend/app/core/runtime_settings.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def apply_governance_runtime_values(structured: dict[str, Any], db: Session) -> tuple[dict[str, Any], list[dict[str, Any]]]:
    """Overlay only settings the Server actually enforces onto a draft."""
    result = json.loads(json.dumps(structured))
    retention = result.setdefault("retention", {})
    features = result.setdefault("optional_features", {})
    changes: list[dict[str, Any]] = []
    for setting_key, (field, label) in GOVERNANCE_RUNTIME_FIELDS.items():
        effective = get_int(setting_key, db)
        previous = retention.get(field)
        retention[field] = effective
        if previous != effective:
            changes.append({
                "setting": setting_key,
                "governance_field": f"retention.{field}",
                "label": label,
                "previous": previous,
                "current": effective,
            })
    # Deployment features are runtime facts, not controller-editable claims.
    # Importing or resaving an older draft must therefore refresh them just as
    # it refreshes the Server-enforced retention periods.
    from app.core.governance_rendering import runtime_feature_state

    for field, effective in runtime_feature_state().items():
        previous = features.get(field)
        features[field] = effective
        if previous != effective:
            changes.append({
                "setting": f"deployment.{field}",
                "governance_field": f"optional_features.{field}",
                "label": field.replace("_", " ").title(),
                "previous": previous,
                "current": effective,
            })
    return result, changes

set_value 

set_value(key: str, value: int, db: Session) -> dict[str, Any]

Persist a runtime override (upsert).

Source code in backend/app/core/runtime_settings.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
def set_value(key: str, value: int, db: Session) -> dict[str, Any]:
    """Persist a runtime override (upsert)."""
    meta = TUNEABLE_SETTINGS.get(key)
    if meta is None:
        raise KeyError(f"Unknown setting: {key}")
    if not (meta["min"] <= value <= meta["max"]):
        raise ValueError(f"{key} must be between {meta['min']} and {meta['max']}")
    from app.models.server_setting import ServerSetting
    row = db.query(ServerSetting).filter(ServerSetting.key == key).first()
    if row:
        row.value = str(value)
    else:
        db.add(ServerSetting(key=key, value=str(value)))
    # The governance overlay must see this exact value even when the caller's
    # session has not otherwise issued a flushing query yet.
    db.flush()
    impact = _sync_governance_draft(db) if key in GOVERNANCE_RUNTIME_FIELDS else governance_impact(db)
    db.commit()
    return impact

Security

security

Security helpers - session auth, current user dependency. Passkey-only: no password hashing needed for regular auth.

get_current_user 

get_current_user(request: Request, db: Session = Depends(get_db)) -> User

Get the current authenticated user and refresh session activity.

Source code in backend/app/core/security.py 
87
88
89
90
91
92
93
def get_current_user(
    request: Request,
    db: Session = Depends(get_db),
) -> User:
    """Get the current authenticated user and refresh session activity."""

    return _get_current_user(request, db, update_last_seen=True)

get_current_user_read_only 

get_current_user_read_only(request: Request, db: Session = Depends(get_db)) -> User

Authenticate without writing session activity to a fenced database.

Source code in backend/app/core/security.py 
 96
 97
 98
 99
100
101
102
def get_current_user_read_only(
    request: Request,
    db: Session = Depends(get_db),
) -> User:
    """Authenticate without writing session activity to a fenced database."""

    return _get_current_user(request, db, update_last_seen=False)

get_current_user_for_commissioning 

get_current_user_for_commissioning(request: Request, db: Session = Depends(get_db)) -> User

Authenticate a root session without lifting the commissioning fence.

Source code in backend/app/core/security.py 
105
106
107
108
109
110
111
112
113
114
115
116
def get_current_user_for_commissioning(
    request: Request,
    db: Session = Depends(get_db),
) -> User:
    """Authenticate a root session without lifting the commissioning fence."""

    return _get_current_user(
        request,
        db,
        update_last_seen=True,
        allow_commissioning=True,
    )

require_commissioning_root 

require_commissioning_root(current_user: User = Depends(get_current_user_for_commissioning)) -> User

Require the authenticated root while the setup wizard is active.

Source code in backend/app/core/security.py 
119
120
121
122
123
124
125
def require_commissioning_root(
    current_user: User = Depends(get_current_user_for_commissioning),
) -> User:
    """Require the authenticated root while the setup wizard is active."""
    if not current_user.is_root_admin:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Root administrator access required")
    return current_user

require_commissioning_root_recent_reauth 

require_commissioning_root_recent_reauth(current_user: User = Depends(require_commissioning_root), db: Session = Depends(get_db)) -> User

Require a setup root whose current session has a recent passkey proof.

Source code in backend/app/core/security.py 
128
129
130
131
132
133
def require_commissioning_root_recent_reauth(
    current_user: User = Depends(require_commissioning_root),
    db: Session = Depends(get_db),
) -> User:
    """Require a setup root whose current session has a recent passkey proof."""
    return ensure_recent_reauth(current_user, db)

require_admin 

require_admin(current_user: User = Depends(get_current_user)) -> User

Dependency: require that the current user is an admin or root admin.

Source code in backend/app/core/security.py 
136
137
138
139
140
141
142
143
def require_admin(current_user: User = Depends(get_current_user)) -> User:
    """Dependency: require that the current user is an admin or root admin."""
    if not current_user.is_root_admin and not current_user.is_admin:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Admin access required",
        )
    return current_user

require_admin_or_issuer 

require_admin_or_issuer(current_user: User = Depends(get_current_user)) -> User

Dependency: require admin, root admin, or issuer.

Source code in backend/app/core/security.py 
146
147
148
149
150
151
152
153
def require_admin_or_issuer(current_user: User = Depends(get_current_user)) -> User:
    """Dependency: require admin, root admin, or issuer."""
    if not current_user.is_root_admin and not current_user.is_admin and not current_user.is_issuer:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Admin or issuer access required",
        )
    return current_user

require_root_or_issuer 

require_root_or_issuer(current_user: User = Depends(get_current_user)) -> User

Dependency: require a root administrator or an issuer account.

Source code in backend/app/core/security.py 
156
157
158
159
160
161
162
163
def require_root_or_issuer(current_user: User = Depends(get_current_user)) -> User:
    """Dependency: require a root administrator or an issuer account."""
    if not current_user.is_root_admin and not current_user.is_issuer:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Root administrator or issuer access required",
        )
    return current_user

require_same_event 

require_same_event(target_user: User, current_user: User) -> None

Raise 403 if current issuer-only user doesn't share event with target.

Source code in backend/app/core/security.py 
171
172
173
174
175
176
177
178
def require_same_event(target_user: User, current_user: User) -> None:
    """Raise 403 if current issuer-only user doesn't share event with target."""
    if _is_issuer_only(current_user):
        if current_user.event_id is None or target_user.event_id != current_user.event_id:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="No access to users from other events",
            )

require_user_management_access 

require_user_management_access(target_user: User, current_user: User) -> None

Enforce account hierarchy and issuer event scope for user management.

Only root may manage another root, global administrator, or issuer account. Issuers may manage ordinary users only within their own event.

Source code in backend/app/core/security.py 
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
def require_user_management_access(target_user: User, current_user: User) -> None:
    """Enforce account hierarchy and issuer event scope for user management.

    Only root may manage another root, global administrator, or issuer account.
    Issuers may manage ordinary users only within their own event.
    """
    if target_user.is_root_admin:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Cannot manage the root admin account",
        )
    if (
        target_user.is_admin or target_user.is_issuer
    ) and not current_user.is_root_admin:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Only root admin can manage privileged accounts",
        )
    require_same_event(target_user, current_user)

require_event_access 

require_event_access(event_id: int, current_user: User, db: Session) -> Event

Return an event when the current user may access it.

Root and global admins may access every event. Issuers, editors, and viewers must have an exact, non-null event assignment.

Source code in backend/app/core/security.py 
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
def require_event_access(event_id: int, current_user: User, db: Session) -> Event:
    """Return an event when the current user may access it.

    Root and global admins may access every event. Issuers, editors, and
    viewers must have an exact, non-null event assignment.
    """
    event = db.query(Event).filter(Event.id == event_id).first()
    if event is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Event not found")
    if current_user.is_root_admin or current_user.is_admin:
        return event
    if current_user.event_id is None or current_user.event_id != event_id:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="No access to this event",
        )
    return event

require_root_admin 

require_root_admin(current_user: User = Depends(get_current_user)) -> User

Dependency: require root admin.

Source code in backend/app/core/security.py 
221
222
223
224
225
226
227
228
def require_root_admin(current_user: User = Depends(get_current_user)) -> User:
    """Dependency: require root admin."""
    if not current_user.is_root_admin:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Root admin access required",
        )
    return current_user

require_root_admin_read_only 

require_root_admin_read_only(current_user: User = Depends(get_current_user_read_only)) -> User

Require root access without mutating session state.

Source code in backend/app/core/security.py 
231
232
233
234
235
236
237
238
239
240
241
def require_root_admin_read_only(
    current_user: User = Depends(get_current_user_read_only),
) -> User:
    """Require root access without mutating session state."""

    if not current_user.is_root_admin:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Root admin access required",
        )
    return current_user

ensure_recent_reauth 

ensure_recent_reauth(current_user: User, db: Session) -> User

Require a recent passkey verification on the current session.

Source code in backend/app/core/security.py 
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
def ensure_recent_reauth(current_user: User, db: Session) -> User:
    """Require a recent passkey verification on the current session."""
    auth_session = getattr(current_user, "_auth_session", None)
    if auth_session is None:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Re-authentication required",
        )
    reauth_at = auth_session.reauth_at
    if reauth_at is None:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Re-authentication required",
        )
    if reauth_at.tzinfo is None:
        reauth_at = reauth_at.replace(tzinfo=timezone.utc)
    window = runtime_settings.get_int("reauth_window_minutes", db)
    if datetime.now(timezone.utc) > reauth_at + timedelta(minutes=window):
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Re-authentication required",
        )
    return current_user

require_recent_reauth 

require_recent_reauth(current_user: User = Depends(require_admin_or_issuer), db: Session = Depends(get_db)) -> User

Require a recently re-authenticated global admin or issuer.

Source code in backend/app/core/security.py 
269
270
271
272
273
274
def require_recent_reauth(
    current_user: User = Depends(require_admin_or_issuer),
    db: Session = Depends(get_db),
) -> User:
    """Require a recently re-authenticated global admin or issuer."""
    return ensure_recent_reauth(current_user, db)

require_admin_recent_reauth 

require_admin_recent_reauth(current_user: User = Depends(require_admin), db: Session = Depends(get_db)) -> User

Require a recently re-authenticated root or global admin.

Source code in backend/app/core/security.py 
277
278
279
280
281
282
def require_admin_recent_reauth(
    current_user: User = Depends(require_admin),
    db: Session = Depends(get_db),
) -> User:
    """Require a recently re-authenticated root or global admin."""
    return ensure_recent_reauth(current_user, db)

require_root_recent_reauth 

require_root_recent_reauth(current_user: User = Depends(require_root_admin), db: Session = Depends(get_db)) -> User

Dependency: require root admin with recent re-authentication.

Source code in backend/app/core/security.py 
285
286
287
288
289
290
def require_root_recent_reauth(
    current_user: User = Depends(require_root_admin),
    db: Session = Depends(get_db),
) -> User:
    """Dependency: require root admin with recent re-authentication."""
    return ensure_recent_reauth(current_user, db)

create_default_admin 

create_default_admin(db: Session) -> User

Create root admin user (passkey-only) if it doesn't exist.

Source code in backend/app/core/security.py 
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
def create_default_admin(db: Session) -> User:
    """Create root admin user (passkey-only) if it doesn't exist."""
    root = db.query(User).filter(User.is_root_admin == True).first()
    if not root:
        root = User(
            username="root.admin",
            display_name="Root Administrator",
            email="root-admin",
            is_root_admin=True,
            is_admin=True,
            is_activated=True,
        )
        db.add(root)
        db.commit()
        db.refresh(root)
        print(f"[Startup] Created root admin (passkey-only): {root.username}")
    return root

Sessions

sessions

Server-side session management - ported from MasterplanOptimiserV2 Server. All users authenticate via session cookies (HttpOnly, Secure, SameSite=Lax).

create_session 

create_session(user_id: int, db: Session, ip_address: Optional[str] = None, user_agent: Optional[str] = None, accept_language: Optional[str] = None, is_privileged: bool = False, reauthenticated: bool = False) -> AuthSession

Create a new server-side session.

Source code in backend/app/core/sessions.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def create_session(
    user_id: int,
    db: Session,
    ip_address: Optional[str] = None,
    user_agent: Optional[str] = None,
    accept_language: Optional[str] = None,
    is_privileged: bool = False,
    reauthenticated: bool = False,
) -> AuthSession:
    """Create a new server-side session."""
    ttl_hours = (
        runtime_settings.get_int("session_ttl_hours_admin", db)
        if is_privileged
        else runtime_settings.get_int("session_ttl_hours", db)
    )
    now = datetime.now(timezone.utc)

    raw_token = secrets.token_urlsafe(48)
    session = AuthSession(
        user_id=user_id,
        session_token=_hash_token(raw_token),
        csrf_token=secrets.token_urlsafe(32),
        expires_at=now + timedelta(hours=ttl_hours),
        last_seen_at=now,
        reauth_at=now if reauthenticated else None,
        ip_address=_hash_ip(ip_address),
        user_agent=_coarse_user_agent(user_agent),
        fingerprint=_compute_fingerprint(user_agent, accept_language),
    )
    db.add(session)
    db.commit()
    db.refresh(session)
    # Attach the raw token so callers can set the cookie; the DB only has the hash.
    session._raw_token = raw_token  # type: ignore[attr-defined]
    return session

validate_session 

validate_session(session_token: str, db: Session, user_agent: Optional[str] = None, accept_language: Optional[str] = None, *, update_last_seen: bool = True) -> Optional[AuthSession]

Look up a session token and return it if still valid.

Source code in backend/app/core/sessions.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
def validate_session(
    session_token: str,
    db: Session,
    user_agent: Optional[str] = None,
    accept_language: Optional[str] = None,
    *,
    update_last_seen: bool = True,
) -> Optional[AuthSession]:
    """Look up a session token and return it if still valid."""
    token_hash = _hash_token(session_token)
    session = (
        db.query(AuthSession)
        .filter(
            AuthSession.session_token == token_hash,
            AuthSession.revoked_at.is_(None),
        )
        .first()
    )
    if session is None:
        return None

    now = datetime.now(timezone.utc)

    expires_at = session.expires_at
    if expires_at.tzinfo is None:
        expires_at = expires_at.replace(tzinfo=timezone.utc)
    if now > expires_at:
        return None

    last_seen = session.last_seen_at
    if last_seen is not None:
        if last_seen.tzinfo is None:
            last_seen = last_seen.replace(tzinfo=timezone.utc)
        inactivity_limit = last_seen + timedelta(
            minutes=runtime_settings.get_int("session_inactivity_minutes", db)
        )
        if now > inactivity_limit:
            return None

    # Fingerprint validation: reject if stored fingerprint doesn't match
    if session.fingerprint:
        current_fp = _compute_fingerprint(user_agent, accept_language)
        if current_fp is None or not secrets.compare_digest(
            current_fp,
            session.fingerprint,
        ):
            return None

    if update_last_seen:
        session.last_seen_at = now
        db.commit()
    return session

revoke_session 

revoke_session(session_token: str, db: Session) -> bool

Revoke a single session.

Source code in backend/app/core/sessions.py 
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
def revoke_session(session_token: str, db: Session) -> bool:
    """Revoke a single session."""
    token_hash = _hash_token(session_token)
    session = (
        db.query(AuthSession)
        .filter(
            AuthSession.session_token == token_hash,
            AuthSession.revoked_at.is_(None),
        )
        .first()
    )
    if session is None:
        return False
    session.revoked_at = datetime.now(timezone.utc)
    db.commit()
    return True

revoke_all_user_sessions 

revoke_all_user_sessions(user_id: int, db: Session) -> int

Revoke every active session for a user.

Source code in backend/app/core/sessions.py 
216
217
218
219
220
221
222
223
224
225
226
227
228
def revoke_all_user_sessions(user_id: int, db: Session) -> int:
    """Revoke every active session for a user."""
    now = datetime.now(timezone.utc)
    count = (
        db.query(AuthSession)
        .filter(
            AuthSession.user_id == user_id,
            AuthSession.revoked_at.is_(None),
        )
        .update({"revoked_at": now})
    )
    db.commit()
    return count

cleanup_expired_sessions 

cleanup_expired_sessions(db: Session, *, now: datetime | None = None, commit: bool = True) -> int

Delete sessions that are expired or were revoked beyond retention period.

Source code in backend/app/core/sessions.py 
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
def cleanup_expired_sessions(
    db: Session,
    *,
    now: datetime | None = None,
    commit: bool = True,
) -> int:
    """Delete sessions that are expired or were revoked beyond retention period."""
    now = now or datetime.now(timezone.utc)

    # Expired sessions past retention window
    expired_cutoff = now - timedelta(
        days=runtime_settings.get_int("retention_expired_sessions_days", db)
    )
    expired = (
        db.query(AuthSession)
        .filter(AuthSession.expires_at < expired_cutoff)
        .delete()
    )

    # Revoked sessions past retention window
    revoked_cutoff = now - timedelta(
        days=runtime_settings.get_int("retention_revoked_sessions_days", db)
    )
    revoked = (
        db.query(AuthSession)
        .filter(
            AuthSession.revoked_at.isnot(None),
            AuthSession.revoked_at < revoked_cutoff,
        )
        .delete()
    )

    if commit:
        db.commit()
    return expired + revoked

Snapshots

snapshots

Snapshot helpers - create / deduplicate / prune publish snapshots.

Used by publish.py (after data insertion) and history.py (rollback).

create_snapshot 

create_snapshot(event: Event, db: Session, source: str = 'desktop') -> Optional[PublishSnapshot]

Snapshot the current published state for an event.

Returns the new snapshot, or None if: - There are no existing tasks (nothing to archive) - Any existing snapshot already has the same content hash (dedup)

Source code in backend/app/core/snapshots.py 
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
def create_snapshot(
    event: Event,
    db: Session,
    source: str = "desktop",
) -> Optional[PublishSnapshot]:
    """Snapshot the current published state for an event.

    Returns the new snapshot, or None if:
    - There are no existing tasks (nothing to archive)
    - Any existing snapshot already has the same content hash (dedup)
    """
    tasks = (
        db.query(PublishedTask)
        .filter(PublishedTask.event_id == event.id)
        .order_by(PublishedTask.sort_order, PublishedTask.start_datetime)
        .all()
    )
    if not tasks:
        return None

    persons = (
        db.query(PublishedPerson)
        .filter(PublishedPerson.event_id == event.id)
        .order_by(PublishedPerson.last_name, PublishedPerson.first_name)
        .all()
    )
    unavailabilities = (
        db.query(PublishedPersonUnavailability)
        .filter(PublishedPersonUnavailability.event_id == event.id)
        .order_by(
            PublishedPersonUnavailability.working_date,
            PublishedPersonUnavailability.start_datetime,
            PublishedPersonUnavailability.external_person_id,
        )
        .all()
    )

    # Load edits
    task_ids = [t.id for t in tasks]
    edits: List[TaskEdit] = []
    if task_ids:
        edits = db.query(TaskEdit).filter(TaskEdit.task_id.in_(task_ids)).all()
    edits_map: Dict[int, TaskEdit] = {e.task_id: e for e in edits}

    # Build resolved tasks (what users saw) for display
    resolved_tasks = []
    for t in tasks:
        out = _task_to_out(
            t,
            edits_map.get(t.id),
            schedule_day_range=event_schedule_day_range(event.metadata_json),
        )
        if out is not None:
            resolved_tasks.append(out.model_dump())

    # Build raw tasks (original columns) for rollback re-insertion
    raw_tasks = [_serialize_raw_task(t) for t in tasks]
    raw_persons = [_serialize_raw_person(p) for p in persons]
    raw_unavailabilities = [
        {
            "external_person_id": row.external_person_id,
            "working_date": row.working_date,
            "start_datetime": row.start_datetime,
            "end_datetime": row.end_datetime,
        }
        for row in unavailabilities
    ]

    # Edits summary
    edited_task_ids = [e.task_id for e in edits if not e.is_deleted]
    deleted_task_ids = [e.task_id for e in edits if e.is_deleted]

    # Event metadata at this point
    event_meta = {
        "name": event.name,
        "start_date": event.start_date.isoformat() if event.start_date else None,
        "end_date": event.end_date.isoformat() if event.end_date else None,
        "metadata_json": event.metadata_json,
    }

    snapshot_data = {
        "tasks": resolved_tasks,
        "raw_tasks": raw_tasks,
        "persons": raw_persons,
        "unavailabilities": raw_unavailabilities,
        "edits_summary": {
            "edited_task_ids": edited_task_ids,
            "deleted_task_ids": deleted_task_ids,
            "total": len(edits),
        },
        "event_meta": event_meta,
    }

    snapshot_json = json.dumps(snapshot_data, sort_keys=True, default=str)

    # Hash only stable content (raw data + event meta) for dedup.
    # resolved_tasks and edits_summary contain DB auto-increment IDs
    # that change on every delete-and-reinsert cycle.
    hash_data = {
        "raw_tasks": raw_tasks,
        "persons": raw_persons,
        "unavailabilities": raw_unavailabilities,
        "event_meta": event_meta,
    }
    content_hash = hashlib.sha256(
        json.dumps(hash_data, sort_keys=True, default=str).encode()
    ).hexdigest()

    # Deduplication: skip if any existing snapshot has the same content
    existing = (
        db.query(PublishSnapshot.id)
        .filter(
            PublishSnapshot.event_id == event.id,
            PublishSnapshot.content_hash == content_hash,
        )
        .first()
    )
    if existing is not None:
        return None

    # Guard: if all slots are frozen, we can't make room for a new snapshot
    frozen_count = (
        db.query(sa_func.count(PublishSnapshot.id))
        .filter(
            PublishSnapshot.event_id == event.id,
            PublishSnapshot.frozen == True,  # noqa: E712
        )
        .scalar()
    ) or 0
    max_snaps = rt.get_int("max_snapshots_per_event", db)
    if frozen_count >= max_snaps:
        return None

    # Compute next version
    max_version = (
        db.query(sa_func.max(PublishSnapshot.version))
        .filter(PublishSnapshot.event_id == event.id)
        .scalar()
    ) or 0
    next_version = max_version + 1

    snapshot = PublishSnapshot(
        event_id=event.id,
        version=next_version,
        snapshot_json=snapshot_json,
        content_hash=content_hash,
        task_count=len(tasks),
        person_count=len(persons),
        edits_count=len(edits),
        source=source,
    )
    db.add(snapshot)

    # Retention: prune oldest if over limit
    _prune_old_snapshots(event.id, db)

    return snapshot