216 lines
8.6 KiB
Python
216 lines
8.6 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import threading
|
|
import unittest
|
|
import uuid
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from unittest.mock import patch
|
|
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import create_engine, text
|
|
from sqlalchemy.engine import make_url
|
|
|
|
import test_password_recovery as recovery_fixture
|
|
from govoplan_access.backend.api.v1 import auth
|
|
from govoplan_access.backend.auth.principal_cache import principal_summary_cache
|
|
from govoplan_access.backend.db.models import PasswordRecovery
|
|
from govoplan_access.backend.security.passwords import verify_password
|
|
|
|
|
|
@unittest.skipUnless(
|
|
os.environ.get("GOVOPLAN_ACCESS_TEST_POSTGRES_URL"),
|
|
"set GOVOPLAN_ACCESS_TEST_POSTGRES_URL to a disposable PostgreSQL test database",
|
|
)
|
|
class PasswordRecoveryPostgresTests(unittest.TestCase):
|
|
"""Opt-in real transaction races; no application/default database fallback."""
|
|
|
|
def setUp(self) -> None:
|
|
url = make_url(os.environ["GOVOPLAN_ACCESS_TEST_POSTGRES_URL"])
|
|
if url.get_backend_name() != "postgresql" or not url.database:
|
|
raise ValueError("An explicit disposable PostgreSQL database is required")
|
|
self.schema = "access_password_race_" + uuid.uuid4().hex
|
|
self.admin_engine = create_engine(url)
|
|
self.addCleanup(self.admin_engine.dispose)
|
|
with self.admin_engine.begin() as connection:
|
|
connection.execute(text(f'CREATE SCHEMA "{self.schema}"'))
|
|
self.addCleanup(self._drop_schema)
|
|
self.engine = create_engine(
|
|
url,
|
|
connect_args={
|
|
"options": (
|
|
f"-c search_path={self.schema} -c statement_timeout=30000 "
|
|
"-c lock_timeout=15000 -c idle_in_transaction_session_timeout=60000"
|
|
)
|
|
},
|
|
)
|
|
self.addCleanup(self.engine.dispose)
|
|
self.access = recovery_fixture.PasswordRecoveryTests(methodName="runTest")
|
|
self.addCleanup(self._close_fixture)
|
|
# Reuse the synthetic HTTP/security fixture, replacing only its SQLite
|
|
# engine. The production account locks/CAS and request SQL sessions run.
|
|
with patch.object(recovery_fixture, "create_engine", return_value=self.engine):
|
|
self.access.setUp()
|
|
|
|
def _drop_schema(self) -> None:
|
|
# The identifier is generated here, never read from the supplied URL.
|
|
with self.admin_engine.begin() as connection:
|
|
connection.execute(text(f'DROP SCHEMA "{self.schema}" CASCADE'))
|
|
|
|
def _close_fixture(self) -> None:
|
|
try:
|
|
if hasattr(self.access, "client"):
|
|
self.access.client.close()
|
|
if hasattr(self.access, "db"):
|
|
self.access.db.close()
|
|
finally:
|
|
self.access.doCleanups()
|
|
principal_summary_cache.clear()
|
|
|
|
def _post(self, endpoint, payload, headers=None):
|
|
# Separate cookie jars and real per-request SQL sessions in each thread.
|
|
with TestClient(self.access.client.app) as client:
|
|
return client.post(
|
|
"/api/v1/auth/password/" + endpoint,
|
|
json=payload,
|
|
headers=headers or {},
|
|
)
|
|
|
|
def _race(self, endpoint, payloads, *, headers=None, locked_account="person"):
|
|
barrier = threading.Barrier(2)
|
|
real_lock = auth.locked_local_account
|
|
|
|
def synchronized_lock(session, account_id):
|
|
if account_id == locked_account:
|
|
barrier.wait(timeout=10)
|
|
return real_lock(session, account_id)
|
|
|
|
with patch.object(auth, "locked_local_account", side_effect=synchronized_lock):
|
|
with ThreadPoolExecutor(max_workers=2) as executor:
|
|
futures = [
|
|
executor.submit(self._post, endpoint, payload, headers)
|
|
for payload in payloads
|
|
]
|
|
return [future.result(timeout=35) for future in futures]
|
|
|
|
def test_same_recovery_code_has_exactly_one_winner(self):
|
|
issued = self.access.issue()
|
|
self.assertEqual(200, issued.status_code)
|
|
code = issued.json()["recovery_code"]
|
|
passwords = ["Concurrent-recovery-one", "Concurrent-recovery-two"]
|
|
responses = self._race(
|
|
"recover",
|
|
[
|
|
{
|
|
"email": "person@example.test",
|
|
"recovery_code": code,
|
|
"new_password": value,
|
|
}
|
|
for value in passwords
|
|
],
|
|
)
|
|
self.assertEqual([200, 400], sorted(item.status_code for item in responses))
|
|
winner = next(
|
|
index for index, item in enumerate(responses) if item.status_code == 200
|
|
)
|
|
self.access.db.expire_all()
|
|
self.assertTrue(
|
|
verify_password(passwords[winner], self.access.account.password_hash)
|
|
)
|
|
self.assertIsNotNone(self.access.db.query(PasswordRecovery).one().consumed_at)
|
|
|
|
def test_same_old_password_has_exactly_one_winner(self):
|
|
passwords = ["Concurrent-change-one", "Concurrent-change-two"]
|
|
responses = self._race(
|
|
"change",
|
|
[
|
|
{"current_password": "Initial-password", "new_password": value}
|
|
for value in passwords
|
|
],
|
|
headers=self.access.headers(),
|
|
)
|
|
self.assertEqual([200, 401], sorted(item.status_code for item in responses))
|
|
winner = next(
|
|
index for index, item in enumerate(responses) if item.status_code == 200
|
|
)
|
|
self.access.db.expire_all()
|
|
self.assertTrue(
|
|
verify_password(passwords[winner], self.access.account.password_hash)
|
|
)
|
|
|
|
def test_competing_issuance_leaves_exactly_one_usable_code(self):
|
|
responses = self._race(
|
|
"recovery/person",
|
|
[{"current_password": "Owner-password", "identity_verified": True}] * 2,
|
|
headers=self.access.headers(self.access.owner_session),
|
|
locked_account="owner",
|
|
)
|
|
self.assertEqual([200, 200], [item.status_code for item in responses])
|
|
codes = [item.json()["recovery_code"] for item in responses]
|
|
self.assertEqual(2, len(set(codes)))
|
|
self.access.db.expire_all()
|
|
self.assertEqual(
|
|
1,
|
|
self.access.db.query(PasswordRecovery)
|
|
.filter(PasswordRecovery.consumed_at.is_(None))
|
|
.count(),
|
|
)
|
|
self.assertEqual(
|
|
[200, 400], sorted(self.access.recover(code).status_code for code in codes)
|
|
)
|
|
|
|
def test_issuer_password_revocation_wins_paused_redemption(self):
|
|
issued = self.access.issue()
|
|
self.assertEqual(200, issued.status_code)
|
|
code = issued.json()["recovery_code"]
|
|
reached = threading.Event()
|
|
proceed = threading.Event()
|
|
real_authorized = auth.recovery_issuer_authorized
|
|
|
|
def paused_authorization(*args, **kwargs):
|
|
reached.set()
|
|
self.assertTrue(proceed.wait(timeout=35))
|
|
return real_authorized(*args, **kwargs)
|
|
|
|
with patch.object(
|
|
auth, "recovery_issuer_authorized", side_effect=paused_authorization
|
|
):
|
|
with ThreadPoolExecutor(max_workers=1) as executor:
|
|
redemption = executor.submit(
|
|
self._post,
|
|
"recover",
|
|
{
|
|
"email": "person@example.test",
|
|
"recovery_code": code,
|
|
"new_password": "Must-not-replace-password",
|
|
},
|
|
)
|
|
try:
|
|
self.assertTrue(reached.wait(timeout=10))
|
|
changed = self._post(
|
|
"change",
|
|
{
|
|
"current_password": "Owner-password",
|
|
"new_password": "Owner-race-password",
|
|
},
|
|
self.access.headers(self.access.owner_session),
|
|
)
|
|
self.assertEqual(200, changed.status_code)
|
|
finally:
|
|
proceed.set()
|
|
rejected = redemption.result(timeout=35)
|
|
self.assertEqual(400, rejected.status_code)
|
|
self.assertEqual("recovery_invalid", rejected.json()["detail"]["code"])
|
|
self.access.db.expire_all()
|
|
self.assertTrue(
|
|
verify_password("Initial-password", self.access.account.password_hash)
|
|
)
|
|
self.assertTrue(
|
|
verify_password("Owner-race-password", self.access.owner.password_hash)
|
|
)
|
|
self.assertIsNotNone(self.access.db.query(PasswordRecovery).one().consumed_at)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|