diff --git a/apps/api/app/routes/auth.py b/apps/api/app/routes/auth.py index 46dbeb7..9228893 100644 --- a/apps/api/app/routes/auth.py +++ b/apps/api/app/routes/auth.py @@ -58,6 +58,7 @@ GOOGLE_TOKENINFO_URL = "https://oauth2.googleapis.com/tokeninfo" OAUTH_STATE_TTL_SECONDS = 10 * 60 OAUTH_STATE_COOKIE_NAME = "__Host-vignette_oauth_state" DEV_OAUTH_STATE_COOKIE_NAME = "vignette_oauth_state" +GENERIC_ADMIN_ENTRY_PATHS = {"/", "/learn", "/teach", "/login", "/onboarding"} @dataclass(slots=True) @@ -343,6 +344,29 @@ def _safe_next_path(next_path: str | None) -> str: return next_path +def _is_generic_admin_entry_path(next_path: str) -> bool: + path = urlsplit(_safe_next_path(next_path)).path.rstrip("/") or "/" + return path in GENERIC_ADMIN_ENTRY_PATHS + + +def _post_login_next_path( + next_path: str, + *, + email: str, + role: Role, + managed_user: ManagedUser | None, +) -> str: + safe_next = _safe_next_path(next_path) + stored_admin_access = managed_user.admin_access if managed_user is not None else False + if _is_generic_admin_entry_path(safe_next) and has_admin_access( + email, + role.value, + stored_admin_access, + ): + return "/admin" + return safe_next + + def _url_origin(value: str | None) -> str | None: if not value: return None @@ -951,7 +975,13 @@ async def callback( ) return _oauth_callback_error("inactive_user", request) - response = RedirectResponse(_frontend_url(stored.next_path, request), status_code=302) + next_path = _post_login_next_path( + stored.next_path, + email=email, + role=role, + managed_user=managed_user, + ) + response = RedirectResponse(_frontend_url(next_path, request), status_code=302) _set_session_cookie(response, sid) _delete_oauth_state_cookie(response) return response @@ -1010,7 +1040,13 @@ async def saml_acs(request: Request) -> RedirectResponse: except InactiveUserError: return _frontend_login_redirect("inactive_user", request) - response = RedirectResponse(_frontend_url(stored.next_path, request), status_code=302) + next_path = _post_login_next_path( + stored.next_path, + email=email, + role=role, + managed_user=managed_user, + ) + response = RedirectResponse(_frontend_url(next_path, request), status_code=302) _set_session_cookie(response, sid) return response diff --git a/apps/api/app/test_auth_providers.py b/apps/api/app/test_auth_providers.py index 507c569..6902689 100644 --- a/apps/api/app/test_auth_providers.py +++ b/apps/api/app/test_auth_providers.py @@ -97,6 +97,40 @@ def _fixture_saml_response( return base64.b64encode(xml.encode("utf-8")).decode("ascii") +def _managed_user( + *, + email: str, + role: str = "learner", + admin_access: bool = False, +) -> auth_sessions.ManagedUser: + return auth_sessions.ManagedUser( + user_id=auth_sessions.user_id_from_email(email), + email=email, + display_name=email, + role=role, + admin_access=admin_access, + account_status="approved", + cohort_ids=[], + affiliation="", + legal_name="", + department="", + grade_level="", + phone="", + contact_address="", + nickname="", + self_introduction="", + avatar_url="", + consent_at=None, + profile_completed_at=1.0, + terms_agreed_at=1.0, + privacy_agreed_at=1.0, + terms_version="", + privacy_version="", + created_at=1.0, + last_seen_at=1.0, + ) + + class AuthProviderScaffoldTest(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self) -> None: auth_routes._oauth_states.clear() @@ -695,6 +729,52 @@ class AuthProviderScaffoldTest(unittest.IsolatedAsyncioTestCase): self.assertNotIn("SAMLResponse", cookie_blob) self.assertNotIn(relay_state, auth_routes._saml_states) + async def test_saml_acs_sends_admin_entitled_generic_next_to_admin(self) -> None: + relay_state = "relay-state" + auth_routes._saml_states[relay_state] = auth_routes.SamlState( + request_id="_request", + next_path="/teach", + created_at=1_800_000_000.0, + ) + request = _form_request( + "/auth/saml/acs", + { + "RelayState": relay_state, + "SAMLResponse": _fixture_saml_response( + email="operator@hs.ac.kr", + display_name="Operator", + role="teacher", + ), + }, + ) + managed = _managed_user(email="operator@hs.ac.kr", role="learner", admin_access=True) + + create_session_mock = AsyncMock(return_value=("opaque-session", object())) + with ( + patched_settings( + auth_saml_enabled=True, + saml_sp_entity_id="https://api-vignette.chanpaca.net/auth/saml/metadata", + saml_sso_url="https://sso.hs.ac.kr/idp/profile/SAML2/Redirect/SSO", + saml_x509_cert_fingerprint="", + frontend_base_url="https://vignette.test", + environment="dev", + ), + patch.object(auth_routes, "get_managed_user_by_email", AsyncMock(return_value=managed)), + patch.object(auth_routes, "create_session", create_session_mock), + ): + response = await auth_routes.saml_acs(request) + + self.assertEqual(response.status_code, 302) + self.assertEqual(response.headers["location"], "https://vignette.test/admin") + create_session_mock.assert_awaited_once_with( + email="operator@hs.ac.kr", + display_name="Operator", + role="learner", + cohort_ids=[], + external_id="saml:operator@hs.ac.kr", + ) + self.assertNotIn(relay_state, auth_routes._saml_states) + async def test_saml_acs_rejects_bad_relay_state(self) -> None: request = _form_request( "/auth/saml/acs", @@ -929,6 +1009,90 @@ class AuthProviderScaffoldTest(unittest.IsolatedAsyncioTestCase): self.assertNotIn("browser-must-not-see-this", cookie_blob) self.assertNotIn(state, auth_routes._oauth_states) + async def test_google_callback_sends_admin_entitled_generic_next_to_admin(self) -> None: + state = "state-token" + auth_routes._oauth_states[state] = auth_routes.OAuthState( + code_verifier="verifier", + next_path="/learn", + created_at=1_800_000_000.0, + ) + managed = _managed_user(email="operator@hs.ac.kr", role="learner", admin_access=True) + + class FakeResponse: + def __init__(self, status_code: int, payload: dict[str, Any]) -> None: + self.status_code = status_code + self._payload = payload + + def json(self) -> dict[str, Any]: + return self._payload + + class FakeAsyncClient: + def __init__(self, *args: Any, **kwargs: Any) -> None: + pass + + async def __aenter__(self) -> "FakeAsyncClient": + return self + + async def __aexit__(self, exc_type: object, exc: object, tb: object) -> None: + return None + + async def post(self, url: str, **kwargs: Any) -> FakeResponse: + return FakeResponse(200, {"id_token": "id-token"}) + + async def get(self, url: str, **kwargs: Any) -> FakeResponse: + return FakeResponse( + 200, + { + "aud": "google-client", + "iss": "https://accounts.google.com", + "email": "operator@hs.ac.kr", + "email_verified": "true", + "name": "Operator", + "hd": "hs.ac.kr", + }, + ) + + create_session_mock = AsyncMock(return_value=("opaque-session", object())) + with ( + patched_settings( + oauth_google_client_id="google-client", + oauth_google_client_secret="google-secret", + oauth_redirect_uri="https://api-vignette.test/auth/callback", + frontend_base_url="https://vignette.test", + environment="prod", + auth_admin_emails=[], + auth_super_admin_emails=[], + ), + patch.object(auth_routes.httpx, "AsyncClient", FakeAsyncClient), + patch.object(auth_routes, "get_managed_user_by_email", AsyncMock(return_value=managed)), + patch.object(auth_routes, "create_session", create_session_mock), + ): + response = await auth_routes.callback( + _request(), + code="auth-code", + state=state, + oauth_state_cookie=state, + ) + + self.assertEqual(response.status_code, 302) + self.assertEqual(response.headers["location"], "https://vignette.test/admin") + create_session_mock.assert_awaited_once_with( + email="operator@hs.ac.kr", + display_name="Operator", + role="learner", + cohort_ids=[], + external_id="google:operator@hs.ac.kr", + ) + self.assertEqual( + auth_routes._post_login_next_path( + "/learn/session/session-id", + email="operator@hs.ac.kr", + role=Role.LEARNER, + managed_user=managed, + ), + "/learn/session/session-id", + ) + async def test_google_callback_accepts_signed_state_after_process_restart(self) -> None: with patched_settings( environment="prod", diff --git a/docs/guides/architecture.md b/docs/guides/architecture.md index 9beb53a..d9e6d74 100644 --- a/docs/guides/architecture.md +++ b/docs/guides/architecture.md @@ -724,6 +724,9 @@ DB 레벨 이중강제(`04_audit_eval_rls.sql` §5, `app/db.py` `acquire()`): - 프론트의 최초 진입 경로(`initialPathForUser`)는 pending이면 `/pending`, 온보딩 미완료 일반 사용자는 `/onboarding`, 관리자 콘솔 접근권이 있는 사용자는 기본 역할이 learner/teacher여도 `/admin`을 우선한다. 역할 전환용 `roleHomePath`는 그대로 역할별 홈(`/learn`, `/teach`, `/admin`)만 소유한다. +- OAuth/SAML callback은 저장된 `next`가 일반 진입 경로(`/`, `/learn`, `/teach`, `/login`, `/onboarding`)이고 + 로그인 사용자가 관리자 콘솔 접근권을 가지면 `/admin`으로 정규화한다. 단, `/learn/session/...` 같은 깊은 + 링크는 사용자가 의도적으로 연 URL일 수 있으므로 보존한다. - `AUTH_ALLOWED_EMAIL_DOMAINS`는 기본 도메인 게이트다. 단, 슈퍼 관리자/관리자가 `/admin/users`에 미리 만든 정확한 이메일은 도메인 밖이어도 Google/SAML/dev-login의 이메일 검증을 통과한다. 이 예외는 도메인 전체를 열지 않고, provider 로그인 시 기존 `email:<주소>` 관리 row를