from unittest.mock import patch import pytest from django.contrib.auth import get_user_model from django.core.management import call_command from django.http import HttpResponseRedirect from apps.accounts.models import ( EnterpriseGroupMapping, EnterpriseIdentityProvider, ExternalIdentity, UserGroup, ) from apps.audit.models import OperationAuditLog from apps.modelhub.crypto import encrypt_secret pytestmark = pytest.mark.django_db def login_admin(client): call_command("bootstrap_governance") admin = get_user_model().objects.get(username="admin") client.force_login(admin) return admin def oidc_provider(**overrides): values = { "slug": "corporate", "name": "企业统一身份", "provider_type": "oidc", "enabled": True, "issuer": "https://idp.example.com", "client_id": "ticket-platform", "credential_encrypted": encrypt_secret("client-secret"), "scopes": ["openid", "profile", "email"], "claim_mappings": {}, "configuration": {}, } values.update(overrides) return EnterpriseIdentityProvider.objects.create(**values) def test_provider_management_encrypts_credentials_and_exposes_safe_public_list(client): login_admin(client) response = client.post( "/api/identity/providers", data={ "slug": "corp-oidc", "name": "公司 SSO", "provider_type": "oidc", "enabled": True, "issuer": "https://idp.example.com", "client_id": "ticket-platform", "credential": "top-secret-value", "scopes": ["openid", "profile", "email"], }, content_type="application/json", ) assert response.status_code == 200 payload = response.json() assert payload["credential_set"] is True assert "top-secret-value" not in str(payload) provider = EnterpriseIdentityProvider.objects.get(slug="corp-oidc") assert provider.credential_encrypted != "top-secret-value" client.logout() public_response = client.get("/api/identity/auth/providers") assert public_response.status_code == 200 assert public_response.json() == [ { "slug": "corp-oidc", "name": "公司 SSO", "button_label": "使用 公司 SSO 登录", "provider_type": "oidc", } ] def test_provider_rejects_plaintext_secrets_inside_configuration(client): login_admin(client) response = client.post( "/api/identity/providers", data={ "slug": "unsafe", "name": "不安全配置", "provider_type": "ldap", "configuration": { "server_uri": "ldaps://directory.example.com", "bind_password": "must-not-be-plain", }, }, content_type="application/json", ) assert response.status_code == 400 assert not EnterpriseIdentityProvider.objects.filter(slug="unsafe").exists() def test_directory_sync_is_idempotent_and_maps_enterprise_groups(client): login_admin(client) provider = oidc_provider( enabled=False, claim_mappings={ "subject": "employee.id", "username": "employee.account", "email": "employee.mail", "display_name": "employee.name", "groups": "organization.memberships", }, ) support = UserGroup.objects.create(name="一线支持", code="support_l1") EnterpriseGroupMapping.objects.create( provider=provider, external_group="IT-Support", user_group=support, ) url = f"/api/identity/providers/{provider.id}/directory-sync" record = { "subject": "employee-1001", "username": "zhangsan", "email": "zhangsan@example.com", "display_name": "张三", "groups": ["IT-Support"], } first = client.post( url, data={"records": [record]}, content_type="application/json", ) second = client.post( url, data={"records": [{**record, "display_name": "张三（支持）"}]}, content_type="application/json", ) assert first.status_code == 200 assert first.json()["created_count"] == 1 assert second.status_code == 200 assert second.json()["updated_count"] == 1 assert ExternalIdentity.objects.filter(provider=provider).count() == 1 identity = ExternalIdentity.objects.select_related("user").get(provider=provider) assert identity.user.first_name == "张三（支持）" assert identity.user.business_groups.filter(id=support.id).exists() assert not identity.user.has_usable_password() def test_oidc_callback_logs_in_user_and_reuses_external_identity(client): provider = oidc_provider() support = UserGroup.objects.create(name="企业支持组", code="enterprise_support") EnterpriseGroupMapping.objects.create( provider=provider, external_group="support", user_group=support, ) class FakeOIDCClient: def authorize_access_token(self, request): return { "userinfo": { "sub": "stable-subject", "preferred_username": "corp-user", "email": "corp-user@example.com", "name": "企业用户", "groups": ["support"], } } session = client.session session[f"identity_return_to:{provider.slug}"] = "/tickets" session.save() callback = f"/api/identity/auth/oidc/{provider.slug}/callback" with patch( "apps.accounts.identity_api.build_oidc_client", return_value=FakeOIDCClient(), ): first = client.get(callback) client.logout() second = client.get(callback) assert first.status_code == 302 assert first.url.endswith("/#/tickets") assert second.status_code == 302 assert ExternalIdentity.objects.filter(provider=provider).count() == 1 identity = ExternalIdentity.objects.select_related("user").get(provider=provider) assert identity.user.business_groups.filter(id=support.id).exists() assert OperationAuditLog.objects.filter( action="oidc_login", target_id=str(identity.id), ).count() == 2 def test_oidc_start_uses_server_side_client_and_callback_failure_is_safe(client): provider = oidc_provider() class FakeOIDCClient: def authorize_redirect(self, request, redirect_uri): assert redirect_uri.endswith(f"/api/identity/auth/oidc/{provider.slug}/callback") return HttpResponseRedirect("https://idp.example.com/authorize?state=managed") def authorize_access_token(self, request): return {"access_token": "must-never-be-audited"} with patch( "apps.accounts.identity_api.build_oidc_client", return_value=FakeOIDCClient(), ): start = client.get( f"/api/identity/auth/oidc/{provider.slug}/start", {"return_to": "//evil.example.com"}, ) callback = client.get(f"/api/identity/auth/oidc/{provider.slug}/callback") assert start.status_code == 302 assert start.url.startswith("https://idp.example.com/authorize") assert callback.status_code == 302 assert "auth_error=" in callback.url audit = OperationAuditLog.objects.get(action="oidc_callback_failed") assert "must-never-be-audited" not in str(audit.request_snapshot) def test_identity_governance_requires_permission(client): provider = oidc_provider(enabled=False) assert client.get("/api/identity/providers").status_code == 401 assert ( client.post( f"/api/identity/providers/{provider.id}/directory-sync", data={"records": []}, content_type="application/json", ).status_code == 401 )