diff --git a/core/auth.py b/core/auth.py index 4bc9a70dd..66fb6b753 100644 --- a/core/auth.py +++ b/core/auth.py @@ -20,7 +20,6 @@ logger = logging.getLogger(__name__) from core.atomic_io import atomic_write_json as _atomic_write_json # noqa: E402 -from core.middleware import INTERNAL_TOOL_USER # noqa: E402 DEFAULT_PRIVILEGES = { "can_use_agent": True, @@ -49,24 +48,18 @@ ADMIN_PRIVILEGES["allowed_models_restricted"] = False ADMIN_PRIVILEGES["block_all_models"] = False from src.constants import AUTH_FILE, PASSWORD_MIN_LENGTH +from src.owner_identity import RESERVED_AUTH_USERNAMES DEFAULT_AUTH_PATH = AUTH_FILE TOKEN_TTL = 60 * 60 * 24 * 7 # 7 days -# Usernames the auth + middleware layer reserve as internal "synthetic owner" -# sentinels; they must never belong to a real account. The most dangerous is -# "internal-tool": `core.middleware.require_admin` treats any request whose -# `current_user == "internal-tool"` as the in-process tool loopback and grants -# admin, and because the cookie auth path sets `current_user` to the raw -# username, an account literally named "internal-tool" would be silently -# treated as an admin by every `require_admin`-gated route. "api" collides with -# the bearer-token owner-attribution sentinel. "demo"/"system" round out the -# synthetic-owner set the rest of the codebase already special-cases (see -# `_SYNTHETIC_OWNERS` in routes/assistant_routes.py and the matching guards in -# src/task_scheduler.py / routes/research_routes.py) — a real account with one -# of those names would be denied an assistant and inconsistently owner-scoped. -# Refuse to create or rename into any of them so the sentinels can't be -# impersonated. (Keep this in sync with that synthetic-owner set.) -RESERVED_USERNAMES = frozenset({INTERNAL_TOOL_USER, "api", "demo", "system"}) +# Usernames the auth + middleware layer reserves for request sentinels and +# internal storage owners; they must never belong to a real login account. +# "internal-tool" is the most dangerous because `core.middleware.require_admin` +# treats it as the in-process tool loopback. "api" collides with bearer-token +# attribution. "demo"/"system" are synthetic owners already special-cased by +# scheduler/assistant/research paths. The Default/Local owner is a storage +# bucket for explicit auth-disabled no-login mode, not a login username. +RESERVED_USERNAMES = frozenset(RESERVED_AUTH_USERNAMES) def normalize_known_username(users: Dict[str, Any], username: str | None) -> Optional[str]: diff --git a/core/middleware.py b/core/middleware.py index 0e164e35a..51b1f97ca 100644 --- a/core/middleware.py +++ b/core/middleware.py @@ -8,6 +8,8 @@ from fastapi import HTTPException, Request from starlette.middleware.base import BaseHTTPMiddleware from starlette.responses import Response +from src.owner_identity import INTERNAL_TOOL_USER, auth_disabled + # Per-process token that lets the in-app tool layer hit admin-gated # routes via HTTP loopback (the agent's tool calls don't carry the @@ -15,8 +17,6 @@ from starlette.responses import Response # same value from this module. Never persisted or exposed externally. INTERNAL_TOOL_TOKEN = os.environ.get("ODYSSEUS_INTERNAL_TOKEN") or secrets.token_hex(32) INTERNAL_TOOL_HEADER = "X-Odysseus-Internal-Token" -# Pseudo-username on in-process tool-loopback requests; require_admin trusts it and it is reserved. -INTERNAL_TOOL_USER = "internal-tool" def is_cors_preflight(method: str, headers) -> bool: @@ -47,7 +47,7 @@ def require_admin(request: Request): pass auth_mgr = getattr(request.app.state, "auth_manager", None) - if os.getenv("AUTH_ENABLED", "true").lower() == "false": + if auth_disabled(): return if not auth_mgr or not auth_mgr.is_configured: raise HTTPException(403, "Admin only") diff --git a/src/auth_helpers.py b/src/auth_helpers.py index 49f3f01be..d290396c2 100644 --- a/src/auth_helpers.py +++ b/src/auth_helpers.py @@ -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: diff --git a/src/owner_identity.py b/src/owner_identity.py new file mode 100644 index 000000000..3eec83e42 --- /dev/null +++ b/src/owner_identity.py @@ -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 diff --git a/tests/test_owner_identity.py b/tests/test_owner_identity.py new file mode 100644 index 000000000..1f3d159ef --- /dev/null +++ b/tests/test_owner_identity.py @@ -0,0 +1,63 @@ +from types import SimpleNamespace + +import pytest + + +def test_effective_storage_owner_matrix(monkeypatch): + from src.owner_identity import DEFAULT_LOCAL_OWNER, effective_storage_owner + + monkeypatch.delenv("AUTH_ENABLED", raising=False) + assert effective_storage_owner(None) is None + assert effective_storage_owner("") is None + assert effective_storage_owner("alice") == "alice" + for sentinel in ("api", "demo", "system", "internal-tool"): + assert effective_storage_owner(sentinel) is None + assert effective_storage_owner(f" {sentinel.upper()} ") is None + + monkeypatch.setenv("AUTH_ENABLED", "false") + assert effective_storage_owner(None) == DEFAULT_LOCAL_OWNER + assert effective_storage_owner("") == DEFAULT_LOCAL_OWNER + assert effective_storage_owner("admin") == "admin" + for sentinel in ("api", "demo", "system", "internal-tool"): + assert effective_storage_owner(sentinel) is None + + +def test_storage_owner_for_request_uses_api_token_owner(monkeypatch): + from src.auth_helpers import storage_owner_for_request + + monkeypatch.delenv("AUTH_ENABLED", raising=False) + request = SimpleNamespace( + state=SimpleNamespace( + current_user="api", + api_token=True, + api_token_owner="alice", + ) + ) + + assert storage_owner_for_request(request) == "alice" + + +@pytest.mark.parametrize("sentinel", ["api", "demo", "system", "internal-tool"]) +def test_storage_owner_for_request_rejects_request_sentinel(monkeypatch, sentinel): + from src.auth_helpers import storage_owner_for_request + + monkeypatch.delenv("AUTH_ENABLED", raising=False) + request = SimpleNamespace( + state=SimpleNamespace( + current_user=sentinel, + api_token=sentinel == "api", + api_token_owner=None, + ) + ) + + assert storage_owner_for_request(request) is None + + +def test_storage_owner_for_request_uses_default_local_when_auth_disabled(monkeypatch): + from src.auth_helpers import storage_owner_for_request + from src.owner_identity import DEFAULT_LOCAL_OWNER + + monkeypatch.setenv("AUTH_ENABLED", "false") + request = SimpleNamespace(state=SimpleNamespace(current_user=None)) + + assert storage_owner_for_request(request) == DEFAULT_LOCAL_OWNER diff --git a/tests/test_reserved_username_admin_escalation.py b/tests/test_reserved_username_admin_escalation.py index eab5e4539..d70062581 100644 --- a/tests/test_reserved_username_admin_escalation.py +++ b/tests/test_reserved_username_admin_escalation.py @@ -16,8 +16,11 @@ from types import SimpleNamespace import pytest from fastapi import HTTPException +from src.owner_identity import DEFAULT_LOCAL_OWNER from tests.helpers.import_state import clear_module +_RESERVED_NAMES = ["internal-tool", "api", "demo", "system", DEFAULT_LOCAL_OWNER] + def _fresh_auth_manager(tmp_path): # Same import dance as test_security_regressions: drop any cached stub so @@ -30,7 +33,17 @@ def _fresh_auth_manager(tmp_path): @pytest.mark.parametrize( "name", - ["internal-tool", "api", "demo", "system", "INTERNAL-TOOL", " Internal-Tool ", "Api", "SYSTEM"], + [ + "internal-tool", + "api", + "demo", + "system", + DEFAULT_LOCAL_OWNER, + "INTERNAL-TOOL", + " Internal-Tool ", + "Api", + "SYSTEM", + ], ) def test_create_user_rejects_reserved_usernames(tmp_path, name): mgr = _fresh_auth_manager(tmp_path) @@ -45,34 +58,37 @@ def test_create_user_rejects_empty_username(tmp_path): assert "" not in mgr.users -def test_setup_rejects_reserved_admin_username(tmp_path): +@pytest.mark.parametrize("name", _RESERVED_NAMES) +def test_setup_rejects_reserved_admin_username(tmp_path, name): mgr = _fresh_auth_manager(tmp_path) # First-run admin setup funnels through create_user, so it's covered too. - assert mgr.setup("internal-tool", "pw-123456") is False + assert mgr.setup(name, "pw-123456") is False assert mgr.is_configured is False -def test_rename_into_reserved_username_is_blocked(tmp_path): +@pytest.mark.parametrize("name", _RESERVED_NAMES) +def test_rename_into_reserved_username_is_blocked(tmp_path, name): mgr = _fresh_auth_manager(tmp_path) assert mgr.create_user("admin", "pw-123456", is_admin=True) is True assert mgr.create_user("bob", "pw-123456") is True - assert mgr.rename_user("bob", "internal-tool", "admin") is False - assert "internal-tool" not in mgr.users + assert mgr.rename_user("bob", name, "admin") is False + assert name not in mgr.users assert "bob" in mgr.users -def test_legacy_reserved_username_is_removed_on_load(tmp_path): +@pytest.mark.parametrize("name", _RESERVED_NAMES) +def test_legacy_reserved_username_is_removed_on_load(tmp_path, name): auth_path = tmp_path / "auth.json" auth_path.write_text( - '{"users": {"internal-tool": {"password_hash": "unused", "is_admin": false}, ' - '"admin": {"password_hash": "unused", "is_admin": true}}}', + '{"users": {"%s": {"password_hash": "unused", "is_admin": false}, ' + '"admin": {"password_hash": "unused", "is_admin": true}}}' % name, encoding="utf-8", ) mgr = _fresh_auth_manager(tmp_path) - assert "internal-tool" not in mgr.users + assert name not in mgr.users assert "admin" in mgr.users - assert "internal-tool" not in auth_path.read_text(encoding="utf-8") + assert name not in auth_path.read_text(encoding="utf-8") def test_legacy_reserved_username_session_cannot_authenticate(tmp_path): @@ -121,15 +137,16 @@ def test_legacy_reserved_username_session_cannot_pass_admin_gate(tmp_path, monke assert exc.value.status_code == 403 -def test_legacy_reserved_single_user_migrates_to_admin(tmp_path): +@pytest.mark.parametrize("name", _RESERVED_NAMES) +def test_legacy_reserved_single_user_migrates_to_admin(tmp_path, name): auth_path = tmp_path / "auth.json" auth_path.write_text( - '{"username": "internal-tool", "password_hash": "unused"}', + '{"username": "%s", "password_hash": "unused"}' % name, encoding="utf-8", ) mgr = _fresh_auth_manager(tmp_path) - assert "internal-tool" not in mgr.users + assert name not in mgr.users assert "admin" in mgr.users assert mgr.is_admin("admin") is True @@ -141,8 +158,8 @@ def test_token_cache_owner_normalization_requires_current_user(): users = {"alice": {}, "admin": {}} assert normalize_known_username(users, " Alice ") == "alice" - assert normalize_known_username(users, "internal-tool") is None - assert normalize_known_username(users, "api") is None + for name in _RESERVED_NAMES: + assert normalize_known_username(users, name) is None assert normalize_known_username(users, "") is None