mirror of
https://github.com/pewdiepie-archdaemon/odysseus.git
synced 2026-11-03 11:42:23 +01:00
This commit is contained in:
+19
-1
@@ -47,7 +47,7 @@ ADMIN_PRIVILEGES["allowed_models_restricted"] = False
|
||||
# backwards for this sentinel.
|
||||
ADMIN_PRIVILEGES["block_all_models"] = False
|
||||
|
||||
from src.constants import AUTH_FILE, PASSWORD_MIN_LENGTH
|
||||
from src.constants import AUTH_FILE, PASSWORD_MIN_LENGTH, USERNAME_MAX_LENGTH
|
||||
from src.owner_identity import RESERVED_AUTH_USERNAMES
|
||||
DEFAULT_AUTH_PATH = AUTH_FILE
|
||||
TOKEN_TTL = 60 * 60 * 24 * 7 # 7 days
|
||||
@@ -62,6 +62,24 @@ TOKEN_TTL = 60 * 60 * 24 * 7 # 7 days
|
||||
RESERVED_USERNAMES = frozenset(RESERVED_AUTH_USERNAMES)
|
||||
|
||||
|
||||
def username_error(username: str | None) -> Optional[str]:
|
||||
"""Why a new username is not acceptable, or None when it is.
|
||||
|
||||
Admin routes carry the name as a URL path segment, where "/" never
|
||||
reaches the handler, so such an account could not be managed afterwards.
|
||||
"""
|
||||
name = (username or "").strip()
|
||||
if not name:
|
||||
return "Username is required"
|
||||
if len(name) > USERNAME_MAX_LENGTH:
|
||||
return f"Username must be at most {USERNAME_MAX_LENGTH} characters"
|
||||
if "/" in name or "\\" in name or any(not c.isprintable() for c in name):
|
||||
return "Username must not contain slashes or control characters"
|
||||
if name in (".", ".."):
|
||||
return "Username is not allowed"
|
||||
return None
|
||||
|
||||
|
||||
def normalize_known_username(users: Dict[str, Any], username: str | None) -> Optional[str]:
|
||||
"""Return a normalized username only when it exists in the auth user map."""
|
||||
key = str(username or "").strip().lower()
|
||||
|
||||
+13
-7
@@ -13,7 +13,7 @@ import re
|
||||
from pathlib import Path
|
||||
|
||||
from core.atomic_io import atomic_write_json, atomic_write_text
|
||||
from core.auth import AuthManager, RESERVED_USERNAMES, SetAdminResult, TOKEN_TTL
|
||||
from core.auth import AuthManager, RESERVED_USERNAMES, SetAdminResult, TOKEN_TTL, username_error
|
||||
from src.constants import DEEP_RESEARCH_DIR, MEMORY_FILE, PASSWORD_MIN_LENGTH, SKILLS_DIR
|
||||
from src.auth_helpers import _auth_disabled
|
||||
from src.rate_limiter import RateLimiter
|
||||
@@ -155,8 +155,9 @@ def setup_auth_routes(auth_manager: AuthManager) -> APIRouter:
|
||||
raise HTTPException(400, "Already configured")
|
||||
if len(body.password) < PASSWORD_MIN_LENGTH:
|
||||
raise HTTPException(400, f"Password must be at least {PASSWORD_MIN_LENGTH} characters")
|
||||
if len(body.username.strip()) < 1:
|
||||
raise HTTPException(400, "Username is required")
|
||||
name_error = username_error(body.username)
|
||||
if name_error:
|
||||
raise HTTPException(400, name_error)
|
||||
if body.username.lower() in RESERVED_USERNAMES:
|
||||
raise HTTPException(403, "Username is reserved")
|
||||
ok = await asyncio.to_thread(auth_manager.setup, body.username, body.password)
|
||||
@@ -175,8 +176,9 @@ def setup_auth_routes(auth_manager: AuthManager) -> APIRouter:
|
||||
raise HTTPException(403, "Registration is disabled. Ask an admin for an account.")
|
||||
if len(body.password) < PASSWORD_MIN_LENGTH:
|
||||
raise HTTPException(400, f"Password must be at least {PASSWORD_MIN_LENGTH} characters")
|
||||
if len(body.username.strip()) < 1:
|
||||
raise HTTPException(400, "Username is required")
|
||||
name_error = username_error(body.username)
|
||||
if name_error:
|
||||
raise HTTPException(400, name_error)
|
||||
if body.username.lower() in RESERVED_USERNAMES:
|
||||
raise HTTPException(403, "Username is reserved")
|
||||
ok = await asyncio.to_thread(auth_manager.create_user, body.username, body.password, is_admin=False)
|
||||
@@ -334,8 +336,9 @@ def setup_auth_routes(auth_manager: AuthManager) -> APIRouter:
|
||||
raise HTTPException(403, "Admin only")
|
||||
if len(body.password) < PASSWORD_MIN_LENGTH:
|
||||
raise HTTPException(400, f"Password must be at least {PASSWORD_MIN_LENGTH} characters")
|
||||
if len(body.username.strip()) < 1:
|
||||
raise HTTPException(400, "Username is required")
|
||||
name_error = username_error(body.username)
|
||||
if name_error:
|
||||
raise HTTPException(400, name_error)
|
||||
if body.username.lower() in RESERVED_USERNAMES:
|
||||
raise HTTPException(403, "Username is reserved")
|
||||
ok = auth_manager.create_user(body.username, body.password, body.is_admin)
|
||||
@@ -385,6 +388,9 @@ def setup_auth_routes(auth_manager: AuthManager) -> APIRouter:
|
||||
new_username = (body.username or "").strip().lower()
|
||||
if not new_username:
|
||||
raise HTTPException(400, "Username required")
|
||||
name_error = username_error(new_username)
|
||||
if name_error:
|
||||
raise HTTPException(400, name_error)
|
||||
if old_username == new_username:
|
||||
return {"ok": True, "username": new_username, "renamed_self": old_username == user}
|
||||
if old_username not in auth_manager.users:
|
||||
|
||||
@@ -175,6 +175,7 @@ AGENT_ISOLATED_TMP_DIRNAME = ".tmp"
|
||||
|
||||
# Auth policy
|
||||
PASSWORD_MIN_LENGTH = 8
|
||||
USERNAME_MAX_LENGTH = 64
|
||||
|
||||
# Default parameters
|
||||
DEFAULT_TEMPERATURE = 1.0
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
"""New usernames must be usable as a URL path segment and not carry control characters."""
|
||||
|
||||
import asyncio
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
from fastapi import HTTPException
|
||||
|
||||
from tests.test_auth_policy import _auth_module, _make_manager, _setup_endpoint, _signup_endpoint
|
||||
|
||||
BAD = ["a/b", "../x", "a\\b", "tab\tname", "new\nline", "..", ".", "u" * 65]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", BAD)
|
||||
def test_username_error_rejects(name):
|
||||
assert _auth_module().username_error(name)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", ["alice", "Bob Smith", "j.doe", "user@example.com", "日本語", "u" * 64])
|
||||
def test_username_error_accepts(name):
|
||||
assert _auth_module().username_error(name) is None
|
||||
|
||||
|
||||
def test_username_error_requires_a_name():
|
||||
assert _auth_module().username_error(" ") == "Username is required"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", ["a/b", "u" * 65])
|
||||
def test_setup_rejects_bad_username(tmp_path, name):
|
||||
mgr = _make_manager(tmp_path)
|
||||
endpoint, SetupRequest = _setup_endpoint(mgr)
|
||||
request = SimpleNamespace(client=SimpleNamespace(host="127.0.0.1"))
|
||||
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
asyncio.run(endpoint(body=SetupRequest(username=name, password="long-enough-pw"), request=request))
|
||||
|
||||
assert exc.value.status_code == 400
|
||||
assert not mgr.is_configured
|
||||
|
||||
|
||||
def test_signup_rejects_bad_username(tmp_path):
|
||||
mgr = _make_manager(tmp_path)
|
||||
mgr.create_user("admin", "admin-password", is_admin=True)
|
||||
mgr.signup_enabled = True
|
||||
endpoint, SignupRequest = _signup_endpoint(mgr)
|
||||
request = SimpleNamespace(client=SimpleNamespace(host="127.0.0.1"))
|
||||
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
asyncio.run(endpoint(body=SignupRequest(username="a/b", password="long-enough-pw"), request=request))
|
||||
|
||||
assert exc.value.status_code == 400
|
||||
assert "a/b" not in mgr.users
|
||||
Reference in New Issue
Block a user