diff --git a/navi/api/routes/auth.py b/navi/api/routes/auth.py index 0f3585f..a9f814a 100644 --- a/navi/api/routes/auth.py +++ b/navi/api/routes/auth.py @@ -65,6 +65,8 @@ platform: str | None = None, ) -> Response: """Redirect to gnexus-auth OAuth authorization endpoint.""" + if not settings.navi_auth_enabled: + raise HTTPException(status_code=503, detail="Authentication is disabled") if not _auth_configured(): raise HTTPException(status_code=503, detail="OAuth is not configured. Set GNAUTH_CLIENT_ID and GNAUTH_CLIENT_SECRET in .env") @@ -108,6 +110,8 @@ error_description: str | None = None, ) -> Response: """Handle OAuth callback from gnexus-auth.""" + if not settings.navi_auth_enabled: + raise HTTPException(status_code=503, detail="Authentication is disabled") if not _auth_configured(): raise HTTPException(status_code=503, detail="OAuth is not configured. Set GNAUTH_CLIENT_ID and GNAUTH_CLIENT_SECRET in .env") diff --git a/navi/auth/_ddl.py b/navi/auth/_ddl.py index 648b3cf..f537a7d 100644 --- a/navi/auth/_ddl.py +++ b/navi/auth/_ddl.py @@ -1,5 +1,7 @@ """Auth DDL — table creation for navi_users and user_auth_sessions.""" +from datetime import datetime, timezone + import asyncpg _DDL = """ @@ -84,7 +86,7 @@ # When auth is disabled, ensure a local anonymous admin user exists so # foreign-key references from sessions/api_tokens stay valid. if not settings.navi_auth_enabled: - now = __import__("datetime").datetime.now(__import__("datetime").timezone.utc) + now = datetime.now(timezone.utc) await conn.execute( """ INSERT INTO navi_users ( diff --git a/navi/auth/deps.py b/navi/auth/deps.py index 0811db6..537168a 100644 --- a/navi/auth/deps.py +++ b/navi/auth/deps.py @@ -104,7 +104,7 @@ # This short-circuits before any cookie/API-token/OAuth lookup. if not settings.navi_auth_enabled: log.debug("auth.resolve_disabled") - return _ANONYMOUS_USER + return _ANONYMOUS_USER.model_copy() if conn is None: log.debug("auth.resolve_no_conn") @@ -457,7 +457,7 @@ async def require_user(user: Annotated[User | None, Depends(get_current_user)]) -> User: if not settings.navi_auth_enabled: - return _ANONYMOUS_USER + return _ANONYMOUS_USER.model_copy() if user is None: raise HTTPException(status_code=401, detail="Authentication required") return user @@ -465,7 +465,7 @@ async def require_admin(user: Annotated[User | None, Depends(get_current_user)]) -> User: if not settings.navi_auth_enabled: - return _ANONYMOUS_USER + return _ANONYMOUS_USER.model_copy() if user is None: raise HTTPException(status_code=401, detail="Authentication required") if user.role != "admin": @@ -478,7 +478,7 @@ user: Annotated[User | None, Depends(get_current_user)], ) -> User: if not settings.navi_auth_enabled: - return _ANONYMOUS_USER + return _ANONYMOUS_USER.model_copy() if user is None: raise HTTPException(status_code=401, detail="Authentication required") if user.role != "admin" and permission not in user.permissions: diff --git a/tests/integration/test_auth_disabled.py b/tests/integration/test_auth_disabled.py index 9f4816c..d4be0e5 100644 --- a/tests/integration/test_auth_disabled.py +++ b/tests/integration/test_auth_disabled.py @@ -77,11 +77,6 @@ ): app.dependency_overrides.pop(getattr(__import__("navi.api.deps", fromlist=[dep]), dep), None) - # Patch WebSocket user resolver at module level so the real function sees disabled auth. - from navi.auth import deps as auth_deps_mod - original_get_current_user_ws = auth_deps_mod.get_current_user_ws - monkeypatch.setattr(auth_deps_mod, "get_current_user_ws", original_get_current_user_ws) - return TestClient(app), store @@ -119,3 +114,17 @@ assert response.status_code == 200 sessions = response.json() assert any(s["session_id"] == session_id for s in sessions) + + +class TestNoAuthOAuthDisabled: + def test_login_rejected_when_auth_disabled(self, no_auth_client): + client, _ = no_auth_client + response = client.get("/auth/login") + assert response.status_code == 503 + assert "disabled" in response.text.lower() + + def test_callback_rejected_when_auth_disabled(self, no_auth_client): + client, _ = no_auth_client + response = client.get("/auth/callback?code=fake&state=fake") + assert response.status_code == 503 + assert "disabled" in response.text.lower() diff --git a/tests/unit/auth/test_deps.py b/tests/unit/auth/test_deps.py index c487f9b..f440ab2 100644 --- a/tests/unit/auth/test_deps.py +++ b/tests/unit/auth/test_deps.py @@ -340,6 +340,37 @@ assert user.role == "user" +# ── WebSocket auth-disabled test ──────────────────────────────────────────── + + +class FakeWebSocket: + """Minimal stand-in for FastAPI WebSocket.""" + + def __init__(self, cookies=None): + self.cookies = cookies or {} + self.state = MagicMock() + self.state.user = None + + +@pytest.mark.asyncio +async def test_get_current_user_ws_returns_anonymous_when_disabled(_auth_env): + _auth_env["settings"].navi_auth_enabled = False + from navi.auth.deps import get_current_user_ws + + ws = FakeWebSocket() + user = await get_current_user_ws(ws) + assert user.id == "anonymous" + assert user.role == "admin" + + +# ── Shared fake used by check_session_access tests ─────────────────────────── + + +class FakeChatSession: + def __init__(self, user_id): + self.user_id = user_id + + # ── require_user / require_admin tests ────────────────────────────────────── @@ -366,7 +397,6 @@ _auth_env["settings"].navi_auth_enabled = False req = FakeRequest() user = await get_current_user(req) - assert user is _ANONYMOUS_USER assert user.id == "anonymous" assert user.role == "admin" @@ -375,14 +405,16 @@ async def test_require_user_returns_anonymous_when_disabled(_auth_env): _auth_env["settings"].navi_auth_enabled = False user = await require_user(None) - assert user is _ANONYMOUS_USER + assert user.id == "anonymous" + assert user.role == "admin" @pytest.mark.asyncio async def test_require_admin_returns_anonymous_when_disabled(_auth_env): _auth_env["settings"].navi_auth_enabled = False user = await require_admin(None) - assert user is _ANONYMOUS_USER + assert user.id == "anonymous" + assert user.role == "admin" @pytest.mark.asyncio @@ -390,7 +422,8 @@ _auth_env["settings"].navi_auth_enabled = False guard = require_permission("navi.sessions.read_all") user = await guard(None) - assert user is _ANONYMOUS_USER + assert user.id == "anonymous" + assert user.role == "admin" @pytest.mark.asyncio @@ -402,14 +435,6 @@ check_session_access(session, user) -# ── check_session_access tests ────────────────────────────────────────────── - - -class FakeChatSession: - def __init__(self, user_id): - self.user_id = user_id - - @pytest.mark.asyncio async def test_check_session_access_owner(): user = User(id="u1", email="u@test.com")