608 lines
24 KiB
Python
608 lines
24 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from datetime import timedelta
|
|
from unittest.mock import patch
|
|
|
|
from fastapi import Depends, FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from govoplan_access.backend.api.v1 import auth
|
|
from govoplan_access.backend.auth.dependencies import (
|
|
AccessApiPrincipalProvider,
|
|
get_api_principal,
|
|
)
|
|
from govoplan_access.backend.auth.principal_cache import principal_summary_cache
|
|
from govoplan_access.backend.db.base import AccessBase
|
|
from govoplan_access.backend.db.models import (
|
|
Account,
|
|
ApiKey,
|
|
AuthSession,
|
|
PasswordRecovery,
|
|
Role,
|
|
SystemRoleAssignment,
|
|
User,
|
|
)
|
|
from govoplan_access.backend.security.api_keys import create_api_key
|
|
from govoplan_access.backend.security.login_throttle import (
|
|
InMemoryLoginAttemptStore,
|
|
LoginThrottle,
|
|
)
|
|
from govoplan_access.backend.security.passwords import hash_password, verify_password
|
|
from govoplan_access.backend.security.password_change import replace_password
|
|
from govoplan_access.backend.security.sessions import create_auth_session
|
|
from govoplan_core.admin.models import SystemSettings
|
|
from govoplan_core.auth import get_api_principal as get_core_api_principal
|
|
from govoplan_core.core.access import CAPABILITY_AUTH_API_PRINCIPAL_PROVIDER
|
|
from govoplan_core.core.modules import ModuleContext
|
|
from govoplan_core.core.registry import PlatformRegistry
|
|
from govoplan_core.core.change_sequence import (
|
|
ChangeSequenceEntry,
|
|
ChangeSequenceRetentionFloor,
|
|
)
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_core.db.session import get_session
|
|
from govoplan_core.security.time import utc_now
|
|
from govoplan_core.settings import settings
|
|
from govoplan_core.tenancy.scope import Tenant, create_scope_tables
|
|
|
|
|
|
class PasswordRecoveryTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine(
|
|
"sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
|
|
)
|
|
create_scope_tables(self.engine)
|
|
AccessBase.metadata.create_all(self.engine)
|
|
Base.metadata.create_all(
|
|
self.engine,
|
|
tables=[
|
|
SystemSettings.__table__,
|
|
ChangeSequenceEntry.__table__,
|
|
ChangeSequenceRetentionFloor.__table__,
|
|
],
|
|
)
|
|
self.factory = sessionmaker(bind=self.engine)
|
|
self.db = self.factory()
|
|
self.tenant = Tenant(id="tenant", slug="tenant", name="Tenant")
|
|
self.account = Account(
|
|
id="person",
|
|
email="person@example.test",
|
|
normalized_email="person@example.test",
|
|
password_hash=hash_password("Initial-password"),
|
|
password_reset_required=True,
|
|
)
|
|
self.owner = Account(
|
|
id="owner",
|
|
email="owner@example.test",
|
|
normalized_email="owner@example.test",
|
|
password_hash=hash_password("Owner-password"),
|
|
)
|
|
self.user = User(
|
|
id="person-member",
|
|
account_id="person",
|
|
tenant_id="tenant",
|
|
email=self.account.email,
|
|
)
|
|
self.owner_user = User(
|
|
id="owner-member",
|
|
account_id="owner",
|
|
tenant_id="tenant",
|
|
email=self.owner.email,
|
|
)
|
|
role = Role(
|
|
id="owner-role",
|
|
tenant_id=None,
|
|
slug="system_owner",
|
|
name="System owner",
|
|
permissions=["system:*"],
|
|
)
|
|
self.owner_assignment = SystemRoleAssignment(
|
|
id="owner-assignment", account_id="owner", role_id=role.id
|
|
)
|
|
self.db.add_all(
|
|
[
|
|
self.tenant,
|
|
self.account,
|
|
self.owner,
|
|
self.user,
|
|
self.owner_user,
|
|
role,
|
|
self.owner_assignment,
|
|
]
|
|
)
|
|
self.db.flush()
|
|
self.current = create_auth_session(self.db, user=self.user)
|
|
self.other = create_auth_session(self.db, user=self.user)
|
|
self.owner_session = create_auth_session(self.db, user=self.owner_user)
|
|
self.key = create_api_key(
|
|
self.db, user=self.user, name="Human automation", scopes=[]
|
|
)
|
|
self.db.commit()
|
|
self.audit = patch.object(auth, "audit_event").start()
|
|
self.addCleanup(patch.stopall)
|
|
patch.object(settings, "auth_local_password_recovery_enabled", True).start()
|
|
patch.object(settings, "auth_principal_cache_enabled", True).start()
|
|
patch.object(settings, "auth_login_throttle_enabled", False).start()
|
|
self.throttle = LoginThrottle(
|
|
InMemoryLoginAttemptStore(),
|
|
identity_limit=50,
|
|
client_limit=100,
|
|
window_seconds=900,
|
|
)
|
|
patch.object(
|
|
auth, "_password_operation_throttle", return_value=self.throttle
|
|
).start()
|
|
app = FastAPI()
|
|
registry = PlatformRegistry()
|
|
registry.configure_capability_context(
|
|
ModuleContext(registry=registry, settings=settings)
|
|
)
|
|
registry.register_capability_factory(
|
|
"access",
|
|
CAPABILITY_AUTH_API_PRINCIPAL_PROVIDER,
|
|
lambda _context: AccessApiPrincipalProvider(),
|
|
)
|
|
app.state.govoplan_registry = registry
|
|
app.include_router(auth.router, prefix="/api/v1")
|
|
|
|
@app.get("/protected")
|
|
def protected(principal=Depends(get_api_principal)):
|
|
return {"account": principal.account_id}
|
|
|
|
@app.get("/core-protected")
|
|
def core_protected(principal=Depends(get_core_api_principal)):
|
|
return {"account": principal.account_id}
|
|
|
|
def session_dependency():
|
|
with self.factory() as session:
|
|
try:
|
|
yield session
|
|
except BaseException:
|
|
session.rollback()
|
|
raise
|
|
|
|
app.dependency_overrides[get_session] = session_dependency
|
|
self.client = TestClient(app)
|
|
principal_summary_cache.clear()
|
|
|
|
def tearDown(self) -> None:
|
|
self.client.close()
|
|
self.db.close()
|
|
self.engine.dispose()
|
|
principal_summary_cache.clear()
|
|
|
|
def headers(self, created=None):
|
|
return {"authorization": "Bearer " + (created or self.current).token}
|
|
|
|
def change(self, **overrides):
|
|
return self.client.post(
|
|
"/api/v1/auth/password/change",
|
|
headers=self.headers(),
|
|
json={
|
|
"current_password": "Initial-password",
|
|
"new_password": "New-secret-password",
|
|
**overrides,
|
|
},
|
|
)
|
|
|
|
def issue(self):
|
|
return self.client.post(
|
|
"/api/v1/auth/password/recovery/person",
|
|
headers=self.headers(self.owner_session),
|
|
json={"current_password": "Owner-password", "identity_verified": True},
|
|
)
|
|
|
|
def recover(self, code, **overrides):
|
|
return self.client.post(
|
|
"/api/v1/auth/password/recover",
|
|
json={
|
|
"email": "person@example.test",
|
|
"recovery_code": code,
|
|
"new_password": "Recovered-password",
|
|
**overrides,
|
|
},
|
|
)
|
|
|
|
def test_first_login_restricts_then_rotates_and_revokes_all_credentials(self):
|
|
login = self.client.post(
|
|
"/api/v1/auth/login",
|
|
json={"email": self.account.email, "password": "Initial-password"},
|
|
)
|
|
self.assertEqual(200, login.status_code, login.text)
|
|
self.assertEqual(
|
|
"change_password", login.json()["user"]["required_auth_action"]
|
|
)
|
|
self.assertEqual([], login.json()["scopes"])
|
|
self.assertEqual(
|
|
403, self.client.get("/protected", headers=self.headers()).status_code
|
|
)
|
|
self.assertEqual(
|
|
403,
|
|
self.client.patch(
|
|
"/api/v1/auth/profile",
|
|
headers=self.headers(),
|
|
json={"display_name": "Escaped"},
|
|
).status_code,
|
|
)
|
|
for path in ("session", "shell"):
|
|
result = self.client.get("/api/v1/auth/" + path, headers=self.headers())
|
|
self.assertEqual(200, result.status_code, result.text)
|
|
self.assertEqual(
|
|
"change_password", result.json()["user"]["required_auth_action"]
|
|
)
|
|
changed = self.change()
|
|
self.assertEqual(200, changed.status_code, changed.text)
|
|
self.assertIsNone(changed.json()["user"]["required_auth_action"])
|
|
self.assertNotEqual(self.current.token, changed.json()["access_token"])
|
|
for old in (self.current.token, self.other.token, self.key.secret):
|
|
self.assertEqual(
|
|
401,
|
|
self.client.get(
|
|
"/protected", headers={"authorization": "Bearer " + old}
|
|
).status_code,
|
|
)
|
|
self.assertEqual(
|
|
200,
|
|
self.client.get(
|
|
"/protected",
|
|
headers={"authorization": "Bearer " + changed.json()["access_token"]},
|
|
).status_code,
|
|
)
|
|
self.db.expire_all()
|
|
self.assertTrue(
|
|
verify_password("New-secret-password", self.account.password_hash)
|
|
)
|
|
self.assertTrue(verify_password("New-secret-password", self.user.password_hash))
|
|
self.assertFalse(self.account.password_reset_required)
|
|
self.assertNotIn("New-secret-password", str(self.audit.call_args_list))
|
|
self.assertNotIn("Initial-password", str(self.audit.call_args_list))
|
|
|
|
def test_failed_current_password_leaves_credentials_and_flag_unchanged(self):
|
|
result = self.change(current_password="wrong-password")
|
|
self.assertEqual(403, result.status_code, result.text)
|
|
self.db.expire_all()
|
|
self.assertTrue(self.account.password_reset_required)
|
|
self.assertIsNone(self.db.get(AuthSession, self.current.model.id).revoked_at)
|
|
self.assertTrue(verify_password("Initial-password", self.account.password_hash))
|
|
self.assertEqual(422, self.change(new_password="Initial-password").status_code)
|
|
|
|
def test_core_api_provider_enforces_current_flag_for_sessions_and_keys(self):
|
|
for headers in (self.headers(), {"x-api-key": self.key.secret}):
|
|
with patch.object(settings, "auth_local_password_recovery_enabled", False):
|
|
self.assertEqual(
|
|
200, self.client.get("/core-protected", headers=headers).status_code
|
|
)
|
|
result = self.client.get("/core-protected", headers=headers)
|
|
self.assertEqual(403, result.status_code)
|
|
self.assertEqual(
|
|
"password_change_required", result.json()["detail"]["code"]
|
|
)
|
|
|
|
def test_validation_responses_do_not_echo_rejected_secrets(self):
|
|
response = self.change(new_password="tiny")
|
|
self.assertEqual(422, response.status_code)
|
|
self.assertNotIn("tiny", response.text)
|
|
self.assertNotIn("Initial-password", response.text)
|
|
self.assertNotIn("input", response.json()["detail"][0])
|
|
|
|
def test_cookie_change_requires_matching_csrf_and_replaces_csrf(self):
|
|
self.client.cookies.set(settings.auth_session_cookie_name, self.current.token)
|
|
self.client.cookies.set(settings.auth_csrf_cookie_name, self.current.csrf_token)
|
|
payload = {
|
|
"current_password": "Initial-password",
|
|
"new_password": "New-secret-password",
|
|
}
|
|
self.assertEqual(
|
|
403,
|
|
self.client.post("/api/v1/auth/password/change", json=payload).status_code,
|
|
)
|
|
response = self.client.post(
|
|
"/api/v1/auth/password/change",
|
|
headers={"x-csrf-token": self.current.csrf_token},
|
|
json=payload,
|
|
)
|
|
self.assertEqual(200, response.status_code, response.text)
|
|
self.assertIn("HttpOnly", response.headers["set-cookie"])
|
|
self.assertNotIn(self.current.csrf_token, response.headers["set-cookie"])
|
|
|
|
def test_opt_in_preserves_advisory_behavior_and_still_allows_change(self):
|
|
with patch.object(settings, "auth_local_password_recovery_enabled", False):
|
|
self.assertEqual(
|
|
200, self.client.get("/protected", headers=self.headers()).status_code
|
|
)
|
|
self.assertIsNone(
|
|
self.client.get("/api/v1/auth/session", headers=self.headers()).json()[
|
|
"user"
|
|
]["required_auth_action"]
|
|
)
|
|
self.assertEqual(409, self.issue().status_code)
|
|
self.assertEqual(409, self.recover("unknown").status_code)
|
|
self.assertEqual(200, self.change().status_code)
|
|
|
|
def test_warm_cache_and_human_api_keys_do_not_bypass_flag(self):
|
|
self.account.password_reset_required = False
|
|
self.db.commit()
|
|
self.assertEqual(
|
|
200, self.client.get("/protected", headers=self.headers()).status_code
|
|
)
|
|
self.assertEqual(
|
|
200,
|
|
self.client.get(
|
|
"/protected", headers={"x-api-key": self.key.secret}
|
|
).status_code,
|
|
)
|
|
self.account.password_reset_required = True
|
|
self.db.commit()
|
|
self.assertEqual(
|
|
403, self.client.get("/protected", headers=self.headers()).status_code
|
|
)
|
|
self.assertEqual(
|
|
403,
|
|
self.client.get(
|
|
"/protected", headers={"x-api-key": self.key.secret}
|
|
).status_code,
|
|
)
|
|
self.assertEqual(
|
|
403,
|
|
self.client.get(
|
|
"/api/v1/auth/session", headers={"x-api-key": self.key.secret}
|
|
).status_code,
|
|
)
|
|
self.assertEqual(
|
|
403,
|
|
self.client.post(
|
|
"/api/v1/auth/password/change",
|
|
headers={"x-api-key": self.key.secret},
|
|
json={
|
|
"current_password": "Initial-password",
|
|
"new_password": "New-secret-password",
|
|
},
|
|
).status_code,
|
|
)
|
|
self.account.password_reset_required = False
|
|
self.db.commit()
|
|
self.assertEqual(
|
|
400,
|
|
self.client.post(
|
|
"/api/v1/auth/password/change",
|
|
headers={"x-api-key": self.key.secret},
|
|
json={
|
|
"current_password": "Initial-password",
|
|
"new_password": "New-secret-password",
|
|
},
|
|
).status_code,
|
|
)
|
|
|
|
def test_external_provider_is_not_forced_or_allowed_to_change_local_password(self):
|
|
for provider in ("oidc", "service_account"):
|
|
with self.subTest(provider=provider):
|
|
self.account.auth_provider = provider
|
|
self.db.commit()
|
|
session_info = self.client.get(
|
|
"/api/v1/auth/session", headers=self.headers()
|
|
)
|
|
self.assertIsNone(session_info.json()["user"]["required_auth_action"])
|
|
self.assertFalse(session_info.json()["user"]["local_password"])
|
|
self.assertEqual(403, self.change().status_code)
|
|
self.assertEqual(
|
|
401,
|
|
self.client.post(
|
|
"/api/v1/auth/login",
|
|
json={
|
|
"email": self.account.email,
|
|
"password": "Initial-password",
|
|
},
|
|
).status_code,
|
|
)
|
|
|
|
def test_recovery_code_is_hashed_one_use_and_revokes_sessions_and_keys(self):
|
|
response = self.issue()
|
|
self.assertEqual(200, response.status_code, response.text)
|
|
code = response.json()["recovery_code"]
|
|
stored = self.db.query(PasswordRecovery).one()
|
|
self.assertNotEqual(code, stored.code_hash)
|
|
self.assertNotIn(code, str(self.audit.call_args_list))
|
|
self.assertEqual("no-store", response.headers["cache-control"])
|
|
self.assertEqual(200, self.recover(code).status_code)
|
|
self.assertEqual(
|
|
400, self.recover(code, new_password="Another-secret").status_code
|
|
)
|
|
self.db.expire_all()
|
|
self.assertTrue(
|
|
verify_password("Recovered-password", self.account.password_hash)
|
|
)
|
|
self.assertFalse(self.account.password_reset_required)
|
|
self.assertEqual(
|
|
0,
|
|
self.db.query(AuthSession)
|
|
.filter(
|
|
AuthSession.account_id == self.account.id,
|
|
AuthSession.revoked_at.is_(None),
|
|
)
|
|
.count(),
|
|
)
|
|
self.assertIsNotNone(self.db.get(ApiKey, self.key.model.id).revoked_at)
|
|
|
|
def test_expiry_supersession_and_current_issuer_authority(self):
|
|
first = self.issue().json()["recovery_code"]
|
|
second = self.issue().json()["recovery_code"]
|
|
self.assertEqual(400, self.recover(first).status_code)
|
|
latest = (
|
|
self.db.query(PasswordRecovery)
|
|
.filter(PasswordRecovery.consumed_at.is_(None))
|
|
.one()
|
|
)
|
|
latest.expires_at = utc_now() - timedelta(seconds=1)
|
|
self.db.commit()
|
|
self.assertEqual(400, self.recover(second).status_code)
|
|
third = self.issue().json()["recovery_code"]
|
|
self.db.delete(self.owner_assignment)
|
|
self.db.commit()
|
|
self.assertEqual(400, self.recover(third).status_code)
|
|
self.assertEqual(403, self.issue().status_code)
|
|
|
|
def test_recovery_rechecks_account_and_membership_state(self):
|
|
code = self.issue().json()["recovery_code"]
|
|
self.user.is_active = False
|
|
self.db.commit()
|
|
self.assertEqual(400, self.recover(code).status_code)
|
|
self.user.is_active = True
|
|
self.account.auth_provider = "oidc"
|
|
self.db.commit()
|
|
self.assertEqual(400, self.recover(code).status_code)
|
|
|
|
def test_recovery_rechecks_tenant_and_issuer_activation(self):
|
|
code = self.issue().json()["recovery_code"]
|
|
self.tenant.is_active = False
|
|
self.db.commit()
|
|
self.assertEqual(400, self.recover(code).status_code)
|
|
self.tenant.is_active = True
|
|
self.owner.is_active = False
|
|
self.db.commit()
|
|
self.assertEqual(400, self.recover(code).status_code)
|
|
|
|
def test_recovery_requires_current_owner_password_and_identity_verification(self):
|
|
response = self.client.post(
|
|
"/api/v1/auth/password/recovery/person",
|
|
headers=self.headers(self.owner_session),
|
|
json={"current_password": "wrong-password", "identity_verified": True},
|
|
)
|
|
self.assertEqual(403, response.status_code)
|
|
response = self.client.post(
|
|
"/api/v1/auth/password/recovery/person",
|
|
headers=self.headers(self.owner_session),
|
|
json={"current_password": "Owner-password", "identity_verified": False},
|
|
)
|
|
self.assertEqual(422, response.status_code)
|
|
self.assertEqual(0, self.db.query(PasswordRecovery).count())
|
|
role = self.db.get(Role, "owner-role")
|
|
role.permissions = ["system:accounts:update"]
|
|
self.db.commit()
|
|
self.assertEqual(403, self.issue().status_code)
|
|
|
|
def test_invalid_recovery_email_or_unchanged_password_does_not_consume_code(self):
|
|
code = self.issue().json()["recovery_code"]
|
|
response = self.recover(code, email="different@example.test")
|
|
self.assertEqual(400, response.status_code)
|
|
self.assertEqual(
|
|
422, self.recover(code, new_password="Initial-password").status_code
|
|
)
|
|
self.assertEqual(200, self.recover(code).status_code)
|
|
|
|
def test_recovery_abuse_is_bounded_even_with_login_throttle_disabled(self):
|
|
throttle = LoginThrottle(
|
|
InMemoryLoginAttemptStore(),
|
|
identity_limit=2,
|
|
client_limit=100,
|
|
window_seconds=900,
|
|
)
|
|
with patch.object(auth, "_password_operation_throttle", return_value=throttle):
|
|
self.assertEqual(400, self.recover("wrong-code").status_code)
|
|
response = self.recover("different-wrong-code")
|
|
self.assertEqual(429, response.status_code)
|
|
self.assertIn("retry-after", response.headers)
|
|
|
|
def test_failed_audit_rolls_back_password_and_session_rotation(self):
|
|
with patch.object(
|
|
auth, "audit_event", side_effect=RuntimeError("audit unavailable")
|
|
):
|
|
with self.assertRaises(RuntimeError):
|
|
self.change()
|
|
self.db.expire_all()
|
|
self.assertTrue(verify_password("Initial-password", self.account.password_hash))
|
|
self.assertIsNone(self.db.get(AuthSession, self.current.model.id).revoked_at)
|
|
|
|
def test_stale_password_authorization_cannot_overwrite_concurrent_change(self):
|
|
from fastapi import HTTPException
|
|
|
|
with self.factory() as competing:
|
|
current = competing.get(Account, self.account.id)
|
|
replace_password(competing, account=current, password="Concurrent-password")
|
|
competing.commit()
|
|
with self.assertRaises(HTTPException) as conflict:
|
|
# This intentionally uses the account state read before the other
|
|
# transaction, modelling a database without row-lock support.
|
|
replace_password(self.db, account=self.account, password="Stale-password")
|
|
self.assertEqual(409, conflict.exception.status_code)
|
|
self.db.rollback()
|
|
self.db.refresh(self.account)
|
|
self.assertTrue(
|
|
verify_password("Concurrent-password", self.account.password_hash)
|
|
)
|
|
|
|
def test_current_password_change_invalidates_outstanding_recovery(self):
|
|
code = self.issue().json()["recovery_code"]
|
|
self.assertEqual(200, self.change().status_code)
|
|
self.assertEqual(400, self.recover(code).status_code)
|
|
|
|
def test_owner_password_change_invalidates_codes_issued_for_other_accounts(self):
|
|
code = self.issue().json()["recovery_code"]
|
|
changed = self.client.post(
|
|
"/api/v1/auth/password/change",
|
|
headers=self.headers(self.owner_session),
|
|
json={
|
|
"current_password": "Owner-password",
|
|
"new_password": "Owner-new-password",
|
|
},
|
|
)
|
|
self.assertEqual(200, changed.status_code, changed.text)
|
|
rejected = self.recover(code)
|
|
self.assertEqual(400, rejected.status_code, rejected.text)
|
|
self.assertEqual("recovery_invalid", rejected.json()["detail"]["code"])
|
|
self.db.expire_all()
|
|
self.assertIsNotNone(self.db.query(PasswordRecovery).one().consumed_at)
|
|
self.assertTrue(verify_password("Initial-password", self.account.password_hash))
|
|
self.assertEqual(
|
|
1, self.audit.call_args.kwargs["details"]["revoked_password_recoveries"]
|
|
)
|
|
|
|
def test_owner_password_recovery_invalidates_codes_issued_for_other_accounts(self):
|
|
code = self.issue().json()["recovery_code"]
|
|
owner_recovery = self.client.post(
|
|
"/api/v1/auth/password/recovery/owner",
|
|
headers=self.headers(self.owner_session),
|
|
json={"current_password": "Owner-password", "identity_verified": True},
|
|
)
|
|
self.assertEqual(200, owner_recovery.status_code, owner_recovery.text)
|
|
recovered = self.recover(
|
|
owner_recovery.json()["recovery_code"],
|
|
email=self.owner.email,
|
|
new_password="Owner-recovered-password",
|
|
)
|
|
self.assertEqual(200, recovered.status_code, recovered.text)
|
|
rejected = self.recover(code)
|
|
self.assertEqual(400, rejected.status_code, rejected.text)
|
|
self.assertEqual("recovery_invalid", rejected.json()["detail"]["code"])
|
|
self.db.expire_all()
|
|
self.assertTrue(verify_password("Initial-password", self.account.password_hash))
|
|
self.assertTrue(
|
|
verify_password("Owner-recovered-password", self.owner.password_hash)
|
|
)
|
|
self.assertEqual(
|
|
0,
|
|
self.db.query(PasswordRecovery)
|
|
.filter(PasswordRecovery.consumed_at.is_(None))
|
|
.count(),
|
|
)
|
|
|
|
def test_flagged_account_can_sign_out_and_requires_csrf_for_cookie_logout(self):
|
|
self.client.cookies.set(settings.auth_session_cookie_name, self.current.token)
|
|
self.client.cookies.set(settings.auth_csrf_cookie_name, self.current.csrf_token)
|
|
self.assertEqual(403, self.client.post("/api/v1/auth/logout").status_code)
|
|
result = self.client.post(
|
|
"/api/v1/auth/logout", headers={"x-csrf-token": self.current.csrf_token}
|
|
)
|
|
self.assertEqual(200, result.status_code)
|
|
self.assertEqual(
|
|
401,
|
|
self.client.get("/api/v1/auth/session", headers=self.headers()).status_code,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|