Add shared credential envelope infrastructure
This commit is contained in:
245
tests/test_credential_envelopes.py
Normal file
245
tests/test_credential_envelopes.py
Normal file
@@ -0,0 +1,245 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import unittest
|
||||
|
||||
from fastapi import HTTPException
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from govoplan_core.auth import ApiPrincipal
|
||||
from govoplan_core.core.access import PrincipalRef
|
||||
from govoplan_core.core.change_sequence import ChangeSequenceEntry
|
||||
from govoplan_core.security.credential_envelopes import (
|
||||
CredentialAccessContext,
|
||||
CredentialEnvelope,
|
||||
CredentialEnvelopeError,
|
||||
create_credential_envelope,
|
||||
credential_envelope_summary,
|
||||
credential_visible_to_context,
|
||||
list_managed_credential_envelopes,
|
||||
resolve_credential_envelope,
|
||||
update_credential_envelope,
|
||||
)
|
||||
from govoplan_core.security.secrets import encrypt_secret
|
||||
from govoplan_core.server.credentials import _target_scope
|
||||
|
||||
|
||||
def _principal(scopes: set[str]) -> ApiPrincipal:
|
||||
return ApiPrincipal(
|
||||
principal=PrincipalRef(
|
||||
account_id="account-1",
|
||||
membership_id="user-1",
|
||||
tenant_id="tenant-1",
|
||||
scopes=frozenset(scopes),
|
||||
),
|
||||
account=object(),
|
||||
user=object(),
|
||||
)
|
||||
|
||||
|
||||
class CredentialEnvelopeTests(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.engine = create_engine("sqlite+pysqlite:///:memory:")
|
||||
ChangeSequenceEntry.__table__.create(self.engine)
|
||||
CredentialEnvelope.__table__.create(self.engine)
|
||||
|
||||
def tearDown(self) -> None:
|
||||
self.engine.dispose()
|
||||
|
||||
def test_inherited_credential_is_filtered_by_module_and_server(self) -> None:
|
||||
row = CredentialEnvelope(
|
||||
id="credential-1",
|
||||
tenant_id="tenant-1",
|
||||
scope_type="tenant",
|
||||
scope_id="tenant-1",
|
||||
name="Shared account",
|
||||
credential_kind="username_password",
|
||||
public_data={"username": "service@example.org"},
|
||||
secret_data_encrypted=encrypt_secret(json.dumps({"password": "not-returned"})),
|
||||
secret_keys=["password"],
|
||||
allowed_modules=["mail"],
|
||||
allowed_server_refs=["mail:server-1"],
|
||||
inherit_to_lower_scopes=True,
|
||||
is_active=True,
|
||||
revision="revision-1",
|
||||
)
|
||||
allowed = CredentialAccessContext(
|
||||
tenant_id="tenant-1",
|
||||
user_id="user-1",
|
||||
target_scope_type="user",
|
||||
target_scope_id="user-1",
|
||||
module_id="mail",
|
||||
server_ref="mail:server-1",
|
||||
)
|
||||
wrong_module = CredentialAccessContext(
|
||||
tenant_id="tenant-1",
|
||||
user_id="user-1",
|
||||
target_scope_type="user",
|
||||
target_scope_id="user-1",
|
||||
module_id="calendar",
|
||||
server_ref="mail:server-1",
|
||||
)
|
||||
wrong_server = CredentialAccessContext(
|
||||
tenant_id="tenant-1",
|
||||
user_id="user-1",
|
||||
target_scope_type="user",
|
||||
target_scope_id="user-1",
|
||||
module_id="mail",
|
||||
server_ref="mail:server-2",
|
||||
)
|
||||
|
||||
self.assertTrue(credential_visible_to_context(row, allowed))
|
||||
self.assertFalse(credential_visible_to_context(row, wrong_module))
|
||||
self.assertFalse(credential_visible_to_context(row, wrong_server))
|
||||
self.assertNotIn("secret_data_encrypted", credential_envelope_summary(row))
|
||||
|
||||
def test_resolution_returns_secret_only_after_access_check(self) -> None:
|
||||
with Session(self.engine) as session:
|
||||
session.add(
|
||||
CredentialEnvelope(
|
||||
id="credential-1",
|
||||
tenant_id="tenant-1",
|
||||
scope_type="tenant",
|
||||
scope_id="tenant-1",
|
||||
name="Shared account",
|
||||
credential_kind="username_password",
|
||||
public_data={"username": "service@example.org"},
|
||||
secret_data_encrypted=encrypt_secret(json.dumps({"password": "secret"})),
|
||||
secret_keys=["password"],
|
||||
allowed_modules=["mail"],
|
||||
allowed_server_refs=[],
|
||||
inherit_to_lower_scopes=True,
|
||||
is_active=True,
|
||||
revision="revision-1",
|
||||
)
|
||||
)
|
||||
session.commit()
|
||||
|
||||
resolved = resolve_credential_envelope(
|
||||
session,
|
||||
credential_id="credential-1",
|
||||
context=CredentialAccessContext(
|
||||
tenant_id="tenant-1",
|
||||
user_id="user-1",
|
||||
target_scope_type="user",
|
||||
target_scope_id="user-1",
|
||||
module_id="mail",
|
||||
),
|
||||
)
|
||||
self.assertEqual(resolved.public_data["username"], "service@example.org")
|
||||
self.assertEqual(resolved.secret_data["password"], "secret")
|
||||
|
||||
with self.assertRaises(CredentialEnvelopeError):
|
||||
resolve_credential_envelope(
|
||||
session,
|
||||
credential_id="credential-1",
|
||||
context=CredentialAccessContext(
|
||||
tenant_id="tenant-2",
|
||||
user_id="user-2",
|
||||
target_scope_type="user",
|
||||
target_scope_id="user-2",
|
||||
module_id="mail",
|
||||
),
|
||||
)
|
||||
|
||||
def test_management_listing_stays_in_tenant_but_ignores_use_site_limits(self) -> None:
|
||||
with Session(self.engine) as session:
|
||||
session.add_all(
|
||||
[
|
||||
CredentialEnvelope(
|
||||
id="tenant-credential",
|
||||
tenant_id="tenant-1",
|
||||
scope_type="tenant",
|
||||
scope_id="tenant-1",
|
||||
name="Tenant restricted",
|
||||
credential_kind="username_password",
|
||||
public_data={},
|
||||
secret_keys=[],
|
||||
allowed_modules=["calendar"],
|
||||
allowed_server_refs=["calendar:source-1"],
|
||||
inherit_to_lower_scopes=True,
|
||||
is_active=True,
|
||||
revision="revision-1",
|
||||
),
|
||||
CredentialEnvelope(
|
||||
id="other-credential",
|
||||
tenant_id="tenant-2",
|
||||
scope_type="tenant",
|
||||
scope_id="tenant-2",
|
||||
name="Other tenant",
|
||||
credential_kind="username_password",
|
||||
public_data={},
|
||||
secret_keys=[],
|
||||
allowed_modules=[],
|
||||
allowed_server_refs=[],
|
||||
inherit_to_lower_scopes=False,
|
||||
is_active=True,
|
||||
revision="revision-2",
|
||||
),
|
||||
]
|
||||
)
|
||||
session.commit()
|
||||
|
||||
rows = list_managed_credential_envelopes(
|
||||
session,
|
||||
tenant_id="tenant-1",
|
||||
)
|
||||
|
||||
self.assertEqual([row.id for row in rows], ["tenant-credential"])
|
||||
|
||||
def test_tenant_credential_permission_cannot_manage_system_scope(self) -> None:
|
||||
tenant_principal = _principal({"access:credential:write"})
|
||||
with self.assertRaises(HTTPException) as raised:
|
||||
_target_scope(
|
||||
tenant_principal,
|
||||
scope_type="system",
|
||||
scope_id=None,
|
||||
write=True,
|
||||
)
|
||||
self.assertEqual(raised.exception.status_code, 403)
|
||||
|
||||
system_principal = _principal({"access:system_credential:write"})
|
||||
self.assertEqual(
|
||||
_target_scope(
|
||||
system_principal,
|
||||
scope_type="system",
|
||||
scope_id=None,
|
||||
write=True,
|
||||
),
|
||||
(None, "system", None),
|
||||
)
|
||||
|
||||
def test_public_data_rejects_secrets_and_kind_changes_require_secret_decision(self) -> None:
|
||||
with Session(self.engine) as session:
|
||||
with self.assertRaisesRegex(CredentialEnvelopeError, "public_data"):
|
||||
create_credential_envelope(
|
||||
session,
|
||||
tenant_id="tenant-1",
|
||||
scope_type="tenant",
|
||||
scope_id="tenant-1",
|
||||
name="Unsafe",
|
||||
credential_kind="username_password",
|
||||
public_data={"nested": [{"clientSecret": "not-public"}]},
|
||||
)
|
||||
|
||||
row = create_credential_envelope(
|
||||
session,
|
||||
tenant_id="tenant-1",
|
||||
scope_type="tenant",
|
||||
scope_id="tenant-1",
|
||||
name="Safe",
|
||||
credential_kind="username_password",
|
||||
public_data={"username": "service@example.org"},
|
||||
secret_data={"password": "secret"},
|
||||
)
|
||||
with self.assertRaisesRegex(CredentialEnvelopeError, "requires replacing"):
|
||||
update_credential_envelope(
|
||||
session,
|
||||
row,
|
||||
credential_kind="token",
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -232,7 +232,7 @@ class DatabaseMigrationTests(unittest.TestCase):
|
||||
|
||||
self.assertEqual(current, configured_migration_heads(url, migration_track="dev"))
|
||||
self.assertEqual(result.current_revision, ",".join(sorted(current)))
|
||||
self.assertIn("0f1e2d3c4b5a", current)
|
||||
self.assertIn("core_credential_envelopes", tables)
|
||||
self.assertIn("audit_outbox_events", tables)
|
||||
self.assertIn("calendar_outbox_operations", tables)
|
||||
self.assertIn("file_connector_profiles", tables)
|
||||
|
||||
Reference in New Issue
Block a user