feat(auth): add IAS ECP SSO [skip ci]
This commit was merged in pull request #19.
This commit is contained in:
Aleksandr Meshchryakov
330
tests/apps/user/test_sso.py
Normal file
330
tests/apps/user/test_sso.py
Normal file
@@ -0,0 +1,330 @@
|
||||
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)
|
||||
Reference in New Issue
Block a user