import os import tempfile from datetime import UTC, datetime, timedelta from urllib.parse import parse_qs, urlencode, urlsplit import jwt from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric import ec from django.apps import apps from django.contrib.auth import get_user_model from django.test import SimpleTestCase, override_settings from rest_framework.test import APIClient, APITestCase from .factories import UserFactory User = get_user_model() @override_settings( SSO_ENABLED=True, SSO_PROVIDER_URL="https://divopk.vniicentr.ru/sso-redirect/", SSO_SERVICE_CODE="fkc", ) class SsoStartViewTest(SimpleTestCase): def test_redirects_to_provider_with_target_and_local_path(self): response = self.client.get( "/api/v1/auth/sso/start/", {"redirect_path": "/open-data?scope=rosatom"}, ) self.assertEqual(response.status_code, 307) redirect = urlsplit(response["Location"]) self.assertEqual( f"{redirect.scheme}://{redirect.netloc}{redirect.path}", "https://divopk.vniicentr.ru/sso-redirect/", ) self.assertEqual( parse_qs(redirect.query), {"target": ["fkc"], "path": ["/open-data?scope=rosatom"]}, ) def test_rejects_absolute_redirect_path(self): response = self.client.get( "/api/v1/auth/sso/start/", {"redirect_path": "https://attacker.example/steal"}, ) self.assertEqual(response.status_code, 400) class SsoCallbackViewTest(APITestCase): callback_url = "/api/v1/auth/sso/" def setUp(self): self.private_key = ec.generate_private_key(ec.SECP256R1()) public_key = self.private_key.public_key().public_bytes( serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo, ) key_file = tempfile.NamedTemporaryFile(delete=False) key_file.write(public_key) key_file.close() self.public_key_path = key_file.name self.settings_override = override_settings( SSO_ENABLED=True, SSO_PROVIDER_URL="https://divopk.vniicentr.ru/sso-redirect/", SSO_SERVICE_CODE="fkc", SSO_ISSUER="dvr", SSO_PUBLIC_KEY_PATH=self.public_key_path, SSO_SESSION_TTL_SECONDS=7200, SSO_TOKEN_MAX_AGE_SECONDS=60, SSO_CLOCK_SKEW_SECONDS=5, SESSION_COOKIE_SECURE=True, SESSION_COOKIE_HTTPONLY=True, SESSION_COOKIE_SAMESITE="Lax", CSRF_COOKIE_SECURE=True, CSRF_COOKIE_SAMESITE="Lax", ) self.settings_override.enable() self.addCleanup(self.settings_override.disable) self.addCleanup(os.unlink, self.public_key_path) def _claims(self, **overrides): issued_at = datetime.now(UTC).replace(microsecond=0) claims = { "token_type": "sso", "iss": "dvr", "iat": int(issued_at.timestamp()), "exp": int((issued_at + timedelta(seconds=60)).timestamp()), "jti": f"{self._testMethodName}-token", "user_id": 2564, "username": "test30", "email": "test30@example.com", "first_name": "Иван", "last_name": "Иванов", "surname": "Иванович", "sso_target": "fkc", "groups": ["Аналитика"], "okpo": "07508753", } claims.update(overrides) return claims def _token(self, **overrides): return self._encode_claims(self._claims(**overrides)) def _encode_claims(self, claims, *, key=None, algorithm="ES256"): return jwt.encode( claims, key or self.private_key, algorithm=algorithm, ) def _post_callback(self, token: str, redirect_path: str = "/open-data"): return self.client.post( self.callback_url, urlencode({"sso_token": token, "redirect_path": redirect_path}), content_type="application/x-www-form-urlencoded", ) def test_valid_token_creates_regular_user_identity_and_session(self): response = self._post_callback(self._token()) self.assertEqual(response.status_code, 303) self.assertEqual(response["Location"], "/open-data") user = User.objects.get(username="test30") self.assertFalse(user.has_usable_password()) self.assertTrue(user.is_verified) self.assertFalse(user.is_staff) self.assertEqual(user.profile.first_name, "Иван") self.assertEqual(user.profile.mid_name, "Иванович") self.assertEqual(user.profile.last_name, "Иванов") identity_model = apps.get_model("user", "SsoIdentity") identity = identity_model.objects.get(user=user) self.assertEqual(identity.issuer, "dvr") self.assertEqual(identity.subject, "2564") self.assertIn("sessionid", response.cookies) self.assertIn("csrftoken", response.cookies) self.assertTrue(response.cookies["sessionid"]["secure"]) self.assertTrue(response.cookies["sessionid"]["httponly"]) self.assertEqual(response.cookies["sessionid"]["samesite"], "Lax") self.assertTrue(response.cookies["csrftoken"]["secure"]) self.assertEqual(response.cookies["csrftoken"]["samesite"], "Lax") self.assertAlmostEqual(self.client.session.get_expiry_age(), 7200, delta=1) self.assertEqual( self.client.session.get("_auth_user_id"), str(user.id), ) current_user_response = self.client.get("/api/v1/users/me/") self.assertEqual(current_user_response.status_code, 200) self.assertEqual(current_user_response.data["username"], "test30") def test_rejects_replayed_jti(self): token = self._token() first_response = self._post_callback(token) replay_response = self._post_callback(token) self.assertEqual(first_response.status_code, 303) self.assertEqual(replay_response.status_code, 403) def test_links_existing_user_only_when_username_and_email_both_match(self): existing_user = UserFactory.create_user( username="test30", email="TEST30@example.com", is_staff=True, ) response = self._post_callback(self._token()) self.assertEqual(response.status_code, 303) self.assertEqual(User.objects.count(), 1) existing_user.refresh_from_db() self.assertTrue(existing_user.is_staff) identity_model = apps.get_model("user", "SsoIdentity") self.assertTrue(identity_model.objects.filter(user=existing_user).exists()) def test_rejects_partial_local_account_collision(self): UserFactory.create_user( username="test30", email="another@example.com", ) response = self._post_callback(self._token()) self.assertEqual(response.status_code, 403) self.assertEqual(User.objects.count(), 1) def test_rejects_invalid_signature_and_algorithm(self): another_private_key = ec.generate_private_key(ec.SECP256R1()) invalid_signature = self._encode_claims( self._claims(jti="invalid-signature"), key=another_private_key, ) invalid_algorithm = self._encode_claims( self._claims(jti="invalid-algorithm"), key="test-only-hmac-secret", algorithm="HS256", ) signature_response = self._post_callback(invalid_signature) algorithm_response = self._post_callback(invalid_algorithm) self.assertEqual(signature_response.status_code, 403) self.assertEqual(algorithm_response.status_code, 403) self.assertFalse(User.objects.exists()) def test_rejects_wrong_issuer_type_and_target(self): invalid_claims = ( {"iss": "other", "jti": "wrong-issuer"}, {"token_type": "access", "jti": "wrong-type"}, {"sso_target": "other", "jti": "wrong-target"}, ) for overrides in invalid_claims: with self.subTest(overrides=overrides): response = self._post_callback(self._token(**overrides)) self.assertEqual(response.status_code, 403) self.assertFalse(User.objects.exists()) def test_rejects_expired_future_and_overlong_tokens(self): now = datetime.now(UTC).replace(microsecond=0) invalid_lifetimes = ( { "iat": int((now - timedelta(seconds=120)).timestamp()), "exp": int((now - timedelta(seconds=60)).timestamp()), "jti": "expired", }, { "iat": int((now + timedelta(seconds=30)).timestamp()), "exp": int((now + timedelta(seconds=60)).timestamp()), "jti": "future", }, { "iat": int(now.timestamp()), "exp": int((now + timedelta(seconds=61)).timestamp()), "jti": "overlong", }, ) for overrides in invalid_lifetimes: with self.subTest(overrides=overrides): response = self._post_callback(self._token(**overrides)) self.assertEqual(response.status_code, 403) self.assertFalse(User.objects.exists()) def test_rejects_missing_required_claim(self): claims = self._claims() claims.pop("email") response = self._post_callback(self._encode_claims(claims)) self.assertEqual(response.status_code, 403) self.assertFalse(User.objects.exists()) def test_rejects_callback_open_redirect_before_creating_user(self): unsafe_paths = ( "//attacker.example/steal", "/\\attacker.example/steal", "/main\nSet-Cookie: stolen=true", ) for index, redirect_path in enumerate(unsafe_paths): with self.subTest(redirect_path=redirect_path): response = self._post_callback( self._token(jti=f"unsafe-redirect-{index}"), redirect_path=redirect_path, ) self.assertEqual(response.status_code, 400) self.assertFalse(User.objects.exists()) def test_rejects_non_form_callback_body(self): response = self.client.post( self.callback_url, {"sso_token": self._token(), "redirect_path": "/main"}, format="json", ) self.assertEqual(response.status_code, 400) self.assertFalse(User.objects.exists()) def test_rejects_inactive_linked_user(self): first_response = self._post_callback(self._token(jti="initial-login")) user = User.objects.get(username="test30") user.is_active = False user.save(update_fields=["is_active"]) second_response = self._post_callback(self._token(jti="inactive-login")) self.assertEqual(first_response.status_code, 303) self.assertEqual(second_response.status_code, 403) user.refresh_from_db() self.assertFalse(user.is_active) def test_session_logout_requires_csrf_and_clears_server_session(self): csrf_client = APIClient(enforce_csrf_checks=True) callback_response = csrf_client.post( self.callback_url, urlencode( { "sso_token": self._token(jti="csrf-login"), "redirect_path": "/main", } ), content_type="application/x-www-form-urlencoded", ) self.assertEqual(callback_response.status_code, 303) missing_csrf_response = csrf_client.post("/api/v1/users/logout/") csrf_token = csrf_client.cookies["csrftoken"].value logout_response = csrf_client.post( "/api/v1/users/logout/", HTTP_X_CSRFTOKEN=csrf_token, ) self.assertEqual(missing_csrf_response.status_code, 403) self.assertEqual(logout_response.status_code, 200) self.assertNotIn("_auth_user_id", csrf_client.session) @override_settings(SSO_ENABLED=False) def test_disabled_sso_returns_service_unavailable(self): start_response = self.client.get("/api/v1/auth/sso/start/") callback_response = self._post_callback(self._token()) self.assertEqual(start_response.status_code, 503) self.assertEqual(callback_response.status_code, 503)