feat(auth): define Default/Local owner contract (#5795)

* feat(auth): define default local owner contract

* test(auth): harden default local owner matrix

---------

Co-authored-by: Alexandre Teixeira <111787685+alteixeira20@users.noreply.github.com>
This commit is contained in:
RaresKeY
2026-08-15 20:27:26 +01:00
committed by GitHub
co-authored by Alexandre Teixeira
parent 9c71948376
commit 0dd70a7556
13 changed files with 250 additions and 45 deletions
+13 -1
View File
@@ -4,6 +4,8 @@ import os
from typing import Optional
from fastapi import Request, HTTPException
from src.owner_identity import auth_disabled, effective_storage_owner
def get_current_user(request: Request) -> Optional[str]:
"""Get current username from request state (set by auth middleware)."""
@@ -56,7 +58,17 @@ def _auth_disabled() -> bool:
"""True when the operator has explicitly turned off auth via .env.
Mirrors the AUTH_ENABLED parse in app.py / core/middleware.py so the
three call sites agree on what "off" means."""
return os.getenv("AUTH_ENABLED", "true").lower() == "false"
return auth_disabled()
def storage_owner_for_request(request: Request) -> Optional[str]:
"""Resolve the storage owner for code paths that need an owner bucket.
This does not replace route authentication. It only gives auth-disabled
no-login mode a stable storage identity instead of writing new data as
legacy NULL/ownerless state.
"""
return effective_storage_owner(effective_user(request))
def require_user(request: Request) -> str:
+56
View File
@@ -0,0 +1,56 @@
"""Shared owner identity constants and helpers."""
from __future__ import annotations
import os
from typing import Optional
DEFAULT_LOCAL_OWNER = "__odysseus_local__"
DEFAULT_LOCAL_OWNER_LABEL = "Local"
INTERNAL_TOOL_USER = "internal-tool"
REQUEST_SENTINEL_OWNERS = frozenset({INTERNAL_TOOL_USER, "api", "demo", "system"})
RESERVED_AUTH_USERNAMES = REQUEST_SENTINEL_OWNERS | {DEFAULT_LOCAL_OWNER}
def auth_disabled() -> bool:
"""Return True only when auth is explicitly disabled by configuration."""
return os.getenv("AUTH_ENABLED", "true").strip().lower() == "false"
def normalize_owner(owner: str | None) -> Optional[str]:
"""Normalize an owner-like value without inventing a fallback identity."""
value = str(owner or "").strip()
return value or None
def owner_key(owner: str | None) -> Optional[str]:
normalized = normalize_owner(owner)
return normalized.lower() if normalized else None
def is_request_sentinel_owner(owner: str | None) -> bool:
return owner_key(owner) in REQUEST_SENTINEL_OWNERS
def effective_storage_owner(owner: str | None, *, auth_is_disabled: bool | None = None) -> Optional[str]:
"""Resolve the owner used for storage writes that need a real bucket.
``None`` still means no authenticated owner when auth is enabled. In the
explicit no-login mode, it resolves to the reserved local owner instead of
conflating local-operator writes with legacy NULL/ownerless rows.
"""
normalized = normalize_owner(owner)
if normalized:
if is_request_sentinel_owner(normalized):
return None
return normalized
disabled = auth_disabled() if auth_is_disabled is None else auth_is_disabled
if disabled:
return DEFAULT_LOCAL_OWNER
return None
def is_default_local_owner(owner: str | None) -> bool:
return owner_key(owner) == DEFAULT_LOCAL_OWNER
+2 -1
View File
@@ -10,6 +10,7 @@ from datetime import datetime, timedelta, timezone
from typing import Any, Awaitable, Callable, Dict, Tuple
from core.auth import RESERVED_USERNAMES
from src.owner_identity import REQUEST_SENTINEL_OWNERS
from src.task_action_policy import (
is_admin_only_task_action,
owner_has_admin_task_privileges,
@@ -2520,7 +2521,7 @@ class TaskScheduler:
# check-ins seeded, which then double-fire alongside the human user's
# check-ins. This was the root cause of the duplicate 'Morning check-in'
# rows we had to manually clean up.
if not owner or owner in RESERVED_USERNAMES:
if not owner or owner in REQUEST_SENTINEL_OWNERS:
logger.info(f"ensure_assistant_defaults: skip synthetic owner {owner!r}")
return
from core.database import SessionLocal, CrewMember, ScheduledTask