331 lines
12 KiB
Python
331 lines
12 KiB
Python
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)
|