246 lines
8.6 KiB
Python
246 lines
8.6 KiB
Python
"""Disposable PostgreSQL contract for durable legacy-auth retirement."""
|
|
|
|
import os
|
|
import unittest
|
|
|
|
from psycopg.errors import CheckViolation
|
|
|
|
from app.agent_auth.exceptions import (
|
|
AgentAuthenticationFailed,
|
|
AgentCredentialRevoked,
|
|
)
|
|
from app.agent_auth.repository import AgentAuthRepository, default_connection_factory
|
|
from app.agent_auth.service import AgentAuthService
|
|
from app.config import settings
|
|
|
|
|
|
class AgentLegacyAuthRetirementPostgresTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
if os.environ.get("MINEROPS_LEGACY_RETIREMENT_DISPOSABLE") != "1":
|
|
raise unittest.SkipTest(
|
|
"legacy-auth retirement requires its disposable PostgreSQL runner"
|
|
)
|
|
cls.case = os.environ["MINEROPS_LEGACY_RETIREMENT_CASE"]
|
|
cls.agent_id = settings.minerops_agent_id
|
|
cls.auth = AgentAuthService(AgentAuthRepository())
|
|
|
|
def row(self):
|
|
connection = default_connection_factory()
|
|
try:
|
|
with connection.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT agent_secret_hash, credential_created_at,
|
|
credential_revoked_at, legacy_auth_disabled_at
|
|
FROM minerops_site_agents
|
|
WHERE agent_id = %s
|
|
""",
|
|
(self.agent_id,),
|
|
)
|
|
return cur.fetchone()
|
|
finally:
|
|
connection.close()
|
|
|
|
def retirement_row(self, agent_id=None):
|
|
connection = default_connection_factory()
|
|
try:
|
|
with connection.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT agent_id, retired_at
|
|
FROM agent_legacy_auth_retirements
|
|
WHERE agent_id = %s
|
|
""",
|
|
(agent_id or self.agent_id,),
|
|
)
|
|
return cur.fetchone()
|
|
finally:
|
|
connection.close()
|
|
|
|
def test_01_migration_backfill_is_selective(self):
|
|
row = self.row()
|
|
self.assertIsNotNone(row)
|
|
if self.case == "legacy":
|
|
self.assertEqual(row, (None, None, None, None))
|
|
self.assertIsNone(self.retirement_row())
|
|
else:
|
|
self.assertIsNotNone(row[3])
|
|
retirement = self.retirement_row()
|
|
self.assertIsNotNone(retirement)
|
|
self.assertEqual(retirement[1], row[3])
|
|
|
|
def test_02_real_repository_authentication_obeys_retirement(self):
|
|
secret = settings.minerops_agent_shared_secret
|
|
if self.case == "legacy":
|
|
self.assertEqual(self.auth.authenticate(None, secret), self.agent_id)
|
|
self.assertEqual(
|
|
self.auth.authenticate(self.agent_id, secret), self.agent_id
|
|
)
|
|
return
|
|
|
|
for agent_id in (None, self.agent_id):
|
|
with self.subTest(agent_id=agent_id), self.assertRaises(
|
|
AgentCredentialRevoked
|
|
):
|
|
self.auth.authenticate(agent_id, secret)
|
|
|
|
def test_03_clearing_credentials_cannot_restore_global_fallback(self):
|
|
if self.case == "legacy":
|
|
self.skipTest("uncredentialed legacy case must remain eligible")
|
|
|
|
connection = default_connection_factory()
|
|
try:
|
|
with connection.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
UPDATE minerops_site_agents
|
|
SET agent_secret_hash = NULL,
|
|
credential_created_at = NULL,
|
|
credential_revoked_at = NULL
|
|
WHERE agent_id = %s
|
|
""",
|
|
(self.agent_id,),
|
|
)
|
|
connection.commit()
|
|
finally:
|
|
connection.close()
|
|
|
|
row = self.row()
|
|
self.assertEqual(row[:3], (None, None, None))
|
|
self.assertIsNotNone(row[3])
|
|
for agent_id in (None, self.agent_id):
|
|
with self.subTest(agent_id=agent_id), self.assertRaises(
|
|
AgentAuthenticationFailed
|
|
):
|
|
self.auth.authenticate(
|
|
agent_id, settings.minerops_agent_shared_secret
|
|
)
|
|
|
|
def test_04_deletion_and_recreation_cannot_restore_fallback(self):
|
|
if self.case == "legacy":
|
|
self.skipTest("uncredentialed legacy case has no retirement history")
|
|
|
|
retirement = self.retirement_row()
|
|
self.assertIsNotNone(retirement)
|
|
|
|
connection = default_connection_factory()
|
|
try:
|
|
with connection.cursor() as cur:
|
|
cur.execute(
|
|
"DELETE FROM minerops_site_agents WHERE agent_id = %s",
|
|
(self.agent_id,),
|
|
)
|
|
connection.commit()
|
|
finally:
|
|
connection.close()
|
|
|
|
self.assertIsNone(self.row())
|
|
self.assertEqual(self.retirement_row(), retirement)
|
|
for agent_id in (None, self.agent_id):
|
|
with self.subTest(state="deleted", agent_id=agent_id), self.assertRaises(
|
|
AgentAuthenticationFailed
|
|
):
|
|
self.auth.authenticate(
|
|
agent_id, settings.minerops_agent_shared_secret
|
|
)
|
|
|
|
connection = default_connection_factory()
|
|
try:
|
|
with connection.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO minerops_site_agents (agent_id, site_name)
|
|
VALUES (%s, 'Recreated')
|
|
""",
|
|
(self.agent_id,),
|
|
)
|
|
connection.commit()
|
|
finally:
|
|
connection.close()
|
|
|
|
recreated = self.row()
|
|
self.assertEqual(recreated[:3], (None, None, None))
|
|
self.assertEqual(recreated[3], retirement[1])
|
|
for agent_id in (None, self.agent_id):
|
|
with self.subTest(state="recreated", agent_id=agent_id), self.assertRaises(
|
|
AgentAuthenticationFailed
|
|
):
|
|
self.auth.authenticate(
|
|
agent_id, settings.minerops_agent_shared_secret
|
|
)
|
|
|
|
def test_05_retirement_ledger_and_projection_are_one_way(self):
|
|
if self.case == "legacy":
|
|
self.skipTest("uncredentialed legacy case has no retirement history")
|
|
|
|
statements = (
|
|
"""
|
|
UPDATE agent_legacy_auth_retirements
|
|
SET retired_at = retired_at + interval '1 second'
|
|
WHERE agent_id = %s
|
|
""",
|
|
"""
|
|
DELETE FROM agent_legacy_auth_retirements
|
|
WHERE agent_id = %s
|
|
""",
|
|
"TRUNCATE agent_legacy_auth_retirements",
|
|
"""
|
|
UPDATE minerops_site_agents
|
|
SET legacy_auth_disabled_at = NULL
|
|
WHERE agent_id = %s
|
|
""",
|
|
)
|
|
for statement in statements:
|
|
connection = default_connection_factory()
|
|
try:
|
|
with self.subTest(statement=" ".join(statement.split())):
|
|
with self.assertRaises(CheckViolation):
|
|
with connection.cursor() as cur:
|
|
if "%s" in statement:
|
|
cur.execute(statement, (self.agent_id,))
|
|
else:
|
|
cur.execute(statement)
|
|
connection.rollback()
|
|
finally:
|
|
connection.close()
|
|
|
|
self.assertIsNotNone(self.retirement_row())
|
|
self.assertIsNotNone(self.row()[3])
|
|
|
|
def test_06_other_legacy_identity_is_not_accidentally_retired(self):
|
|
if self.case == "legacy":
|
|
self.skipTest("cross-identity check runs in the retired case")
|
|
|
|
other_id = self.agent_id + "-untouched"
|
|
connection = default_connection_factory()
|
|
try:
|
|
with connection.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO minerops_site_agents (agent_id, site_name)
|
|
VALUES (%s, 'Untouched')
|
|
""",
|
|
(other_id,),
|
|
)
|
|
connection.commit()
|
|
finally:
|
|
connection.close()
|
|
|
|
self.assertIsNone(self.retirement_row(other_id))
|
|
original_legacy_id = settings.minerops_agent_id
|
|
settings.minerops_agent_id = other_id
|
|
try:
|
|
self.assertEqual(
|
|
self.auth.authenticate(
|
|
other_id, settings.minerops_agent_shared_secret
|
|
),
|
|
other_id,
|
|
)
|
|
finally:
|
|
settings.minerops_agent_id = original_legacy_id
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|