246 lines
9.0 KiB
Python
246 lines
9.0 KiB
Python
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()
|