tking/fastapi-gsap
SHA256
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: wire credential resolver and connectors into broker startup

Browse source 
All connectors registered conditionally based on settings.
CredentialResolver with Entra backend (production) or Stub
backend (dev mode). 15 new tests covering credential resolution,
session lifecycle, orchestrator workflows, and device routing.

Signed-off-by: Tyler King <tking@guildhouse.dev>
This commit is contained in:
Tyler J King 2026-04-14 06:03:57 -04:00
parent 5adc55aff5
commit 4dff879c84
3 changed files with 362 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

37
gsap_broker/routers/connectors.py
View file

@ -21,7 +21,27 @@ _runtime = ConnectorRuntime(registry=_registry)
# Register built-in connectors
_registry.register(EchoConnector())
# Conditionally register Intune connector
# ── Credential resolver (shared by session connectors) ──────────
from gsap_broker.credentials.resolver import CredentialResolver
_credential_resolver = CredentialResolver()
if settings.credential_backend == "stub" or (
settings.credential_backend == "auto" and not settings.entra_client_secret
):
from gsap_broker.credentials.stub_backend import StubCredentialBackend
_credential_resolver.register(StubCredentialBackend())
elif settings.entra_client_secret:
from gsap_broker.credentials.entra_backend import EntraCredentialBackend
_credential_resolver.register(EntraCredentialBackend(
tenant_id=settings.entra_tenant_id,
client_id=settings.entra_client_id,
client_secret=settings.entra_client_secret,
))
# ── Conditionally register connectors ───────────────────────────
# Intune (API-mediated — uses GraphClient, not CredentialResolver)
if settings.intune_enabled and settings.entra_client_secret:
from gsap_broker.intune.graph_client import GraphClient
from gsap_broker.intune.device_cache import DeviceComplianceCache
@ -36,6 +56,21 @@ if settings.intune_enabled and settings.entra_client_secret:
_intune_connector = IntuneConnector(graph_client=_intune_graph, cache=_intune_cache)
_registry.register(_intune_connector)
# Bascule (session-based — uses CredentialResolver)
if settings.bascule_enabled:
from gsap_broker.connectors.bascule import BasculeConnector
_registry.register(BasculeConnector(credential_resolver=_credential_resolver))
# PowerShell (session-based — uses CredentialResolver)
if settings.powershell_enabled:
from gsap_broker.connectors.powershell import PowerShellConnector
_registry.register(PowerShellConnector(credential_resolver=_credential_resolver))
# Ansible (orchestrator — uses CredentialResolver)
if settings.ansible_enabled:
from gsap_broker.connectors.ansible import AnsibleConnector
_registry.register(AnsibleConnector(credential_resolver=_credential_resolver))
class InvokeRequest(BaseModel):
operation: str

10
gsap_broker/settings.py
View file

@ -41,6 +41,16 @@ class Settings(BaseSettings):
intune_compliance_strict: bool = False # reject if no device_id present
intune_compliance_cache_ttl: int = 300 # seconds
# ── Session connectors ──
bascule_enabled: bool = False
powershell_enabled: bool = False
ansible_enabled: bool = False
# ── Credential backend ──
# "auto" | "entra" | "stub"
# auto: use Entra if entra_client_secret is set, else stub
credential_backend: str = "auto"
# Delegation defaults
default_delegation_ttl_minutes: int = 60
default_max_commands: int = 500

316
tests/test_credentials.py Normal file
View file

@ -0,0 +1,316 @@
# Copyright 2026 Guildhouse Dev
# SPDX-License-Identifier: Apache-2.0
"""Tests for credential resolver, backends, and session connectors."""
import pytest
from datetime import datetime, timedelta, UTC
from gsap_broker.credentials.resolver import (
BasculeCredential,
Credential,
CredentialResolver,
CredentialResolutionError,
KerberosCredential,
NoBackendAvailable,
OAuthCredential,
SSHCertCredential,
)
from gsap_broker.credentials.stub_backend import StubCredentialBackend
from gsap_broker.connectors.base import ConnectorContext
from gsap_broker.connectors.bascule import BasculeConnector
from gsap_broker.connectors.powershell import PowerShellConnector
from gsap_broker.connectors.ansible import AnsibleConnector
from gsap_broker.connectors.orchestrator import WorkflowPlan, WorkflowStep
from gsap_broker.routing.device_router import DeviceRouter, UnknownDevice
AC_CONTEXT = {
"gsap_context_id": "test-ac",
"expires_at": (datetime.now(UTC) + timedelta(hours=1)).isoformat(),
"accord": {"template": "test-ops"},
"principal": {"did": "did:web:test/p/alice"},
}
# ── CredentialResolver ────────────────────────────────────────────
@pytest.mark.asyncio
async def test_resolver_routes_to_correct_backend():
"""TEST 2: Resolver routes each credential type to the stub backend."""
resolver = CredentialResolver()
resolver.register(StubCredentialBackend())
for cred_type, expected_cls in [
("bascule_ac", BasculeCredential),
("kerberos", KerberosCredential),
("oauth", OAuthCredential),
("ssh_cert", SSHCertCredential),
]:
cred = await resolver.resolve(cred_type, "target-1", AC_CONTEXT)
assert isinstance(cred, expected_cls)
assert cred.expires_at is not None
assert not cred.expired
@pytest.mark.asyncio
async def test_resolver_rejects_unknown_type():
"""TEST 3: Unknown credential type raises NoBackendAvailable."""
resolver = CredentialResolver()
resolver.register(StubCredentialBackend())
with pytest.raises(NoBackendAvailable) as exc_info:
await resolver.resolve("quantum_key", "target", AC_CONTEXT)
assert "quantum_key" in str(exc_info.value)
@pytest.mark.asyncio
async def test_bascule_credential_preserves_ac():
"""TEST 4: Bascule credential passes AC through."""
resolver = CredentialResolver()
resolver.register(StubCredentialBackend())
cred = await resolver.resolve("bascule_ac", "target-1", AC_CONTEXT)
assert isinstance(cred, BasculeCredential)
assert cred.authorization_context == AC_CONTEXT
@pytest.mark.asyncio
async def test_stub_credentials_not_expired():
"""TEST 5: Stub backend returns credentials that are not expired."""
resolver = CredentialResolver()
resolver.register(StubCredentialBackend())
for cred_type in ["bascule_ac", "kerberos", "oauth", "ssh_cert"]:
cred = await resolver.resolve(cred_type, "target", AC_CONTEXT)
assert not cred.expired
# ── SessionConnector lifecycle ────────────────────────────────────
@pytest.mark.asyncio
async def test_session_connector_success():
"""TEST 6: Full lifecycle — resolve, connect, execute, disconnect."""
resolver = CredentialResolver()
resolver.register(StubCredentialBackend())
connector = BasculeConnector(credential_resolver=resolver)
ctx = ConnectorContext(gsap_context_id="test-ac")
result = await connector.invoke("run-script", {"target": "node-1"}, ctx)
assert result.success
assert result.data["target"] == "node-1"
assert result.data["command"] == "run-script"
assert result.data["stub"] is True
@pytest.mark.asyncio
async def test_session_connector_missing_target():
"""Session connector requires target parameter."""
resolver = CredentialResolver()
resolver.register(StubCredentialBackend())
connector = BasculeConnector(credential_resolver=resolver)
ctx = ConnectorContext(gsap_context_id="test-ac")
result = await connector.invoke("run-script", {}, ctx)
assert not result.success
assert "target required" in result.error
@pytest.mark.asyncio
async def test_session_connector_transport_failure():
"""TEST 7: Transport failure still calls disconnect (cleanup guarantee)."""
from gsap_broker.connectors.session import SessionTransport, SessionConnector
from gsap_broker.credentials.resolver import Credential
disconnect_called = False
class FailingTransport(SessionTransport):
transport_id = "failing"
async def connect(self, target, credential):
pass
async def execute(self, command, params=None):
raise RuntimeError("Transport exploded")
async def disconnect(self):
nonlocal disconnect_called
disconnect_called = True
async def is_alive(self):
return False
class FailingConnector(SessionConnector):
connector_id = "failing"
corpus_entry_cid = "sha256:test"
credential_type = "bascule_ac"
transport_class = FailingTransport
capability_mask = 1
declared_endpoints = []
resolver = CredentialResolver()
resolver.register(StubCredentialBackend())
connector = FailingConnector(credential_resolver=resolver)
ctx = ConnectorContext(gsap_context_id="test-ac")
result = await connector.invoke("crash", {"target": "node-1"}, ctx)
assert not result.success
assert "exploded" in result.error
assert disconnect_called, "disconnect() must be called even on failure"
# ── PowerShell connector ──────────────────────────────────────────
@pytest.mark.asyncio
async def test_powershell_connector_stub():
"""PowerShell connector lifecycle with stubbed transport."""
resolver = CredentialResolver()
resolver.register(StubCredentialBackend())
connector = PowerShellConnector(credential_resolver=resolver)
ctx = ConnectorContext(gsap_context_id="test-ac")
result = await connector.invoke("Get-Process", {"target": "win-srv-01:5986"}, ctx)
assert result.success
assert result.data["transport"] == "psrp"
assert result.data["target"] == "win-srv-01:5986"
# ── OrchestratorConnector (Ansible) ──────────────────────────────
@pytest.mark.asyncio
async def test_orchestrator_all_steps_succeed():
"""TEST 8: All steps succeed → aggregate success."""
resolver = CredentialResolver()
resolver.register(StubCredentialBackend())
connector = AnsibleConnector(credential_resolver=resolver)
ctx = ConnectorContext(gsap_context_id="test-ac")
result = await connector.invoke(
"playbook",
{"playbook": "site.yml", "targets": ["host-1", "host-2"]},
ctx,
)
assert result.success
assert len(result.data["steps"]) == 1
assert result.data["steps"][0]["success"] is True
@pytest.mark.asyncio
async def test_orchestrator_required_step_fails():
"""TEST 9: Required step fails → early termination, partial results."""
from gsap_broker.connectors.orchestrator import OrchestratorConnector, WorkflowPlan, WorkflowStep
resolver = CredentialResolver()
resolver.register(StubCredentialBackend())
class FailingOrchestrator(OrchestratorConnector):
connector_id = "failing-orch"
corpus_entry_cid = "sha256:test"
capability_mask = 1
declared_endpoints = []
async def plan(self, operation, parameters, context):
return WorkflowPlan(steps=[
WorkflowStep(name="step-1", command="ok", required=True),
WorkflowStep(name="step-2", command="fail", required=True),
WorkflowStep(name="step-3", command="never", required=True),
])
async def execute_step(self, step, context):
if step.command == "fail":
return {"success": False, "error": "boom"}
return {"success": True}
connector = FailingOrchestrator(credential_resolver=resolver)
ctx = ConnectorContext(gsap_context_id="test-ac")
result = await connector.invoke("run", {}, ctx)
assert not result.success
assert result.data["failed_at"] == "step-2"
assert len(result.data["completed"]) == 2 # step-1 succeeded, step-2 failed
# step-3 was never executed
# ── DeviceRouter ──────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_router_api_operation_routes_to_intune():
"""TEST 12: API-mediated operations route to Intune."""
from gsap_broker.connectors.registry import ConnectorRegistry
registry = ConnectorRegistry()
router = DeviceRouter(connector_registry=registry)
ctx = ConnectorContext(gsap_context_id="test-ac")
connector_id, op = await router.route("get_compliance", "dev-1", ctx)
assert connector_id == "intune"
@pytest.mark.asyncio
async def test_router_fleet_operation_routes_to_ansible():
"""Fleet operations route to Ansible."""
from gsap_broker.connectors.registry import ConnectorRegistry
registry = ConnectorRegistry()
router = DeviceRouter(connector_registry=registry)
ctx = ConnectorContext(gsap_context_id="test-ac")
connector_id, op = await router.route("playbook", "fleet", ctx)
assert connector_id == "ansible"
@pytest.mark.asyncio
async def test_router_unknown_device_raises():
"""TEST 13: Unknown device raises clear error."""
from gsap_broker.connectors.registry import ConnectorRegistry
registry = ConnectorRegistry()
router = DeviceRouter(connector_registry=registry)
ctx = ConnectorContext(gsap_context_id="test-ac")
with pytest.raises(UnknownDevice):
await router.route("exec", "mystery-host", ctx)
@pytest.mark.asyncio
async def test_router_invoke_missing_connector():
"""Router invoke returns error when connector not registered."""
from gsap_broker.connectors.registry import ConnectorRegistry
registry = ConnectorRegistry()
router = DeviceRouter(connector_registry=registry)
ctx = ConnectorContext(gsap_context_id="test-ac")
result = await router.invoke("get_compliance", "dev-1", {}, ctx)
assert not result.success
assert "not registered" in result.error
# ── Catalog conditional registration ─────────────────────────────
@pytest.mark.asyncio
async def test_connectors_absent_when_disabled():
"""TEST 15: Optional connectors not in catalog when disabled."""
from httpx import AsyncClient, ASGITransport
from gsap_broker.app import app
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
resp = await client.get("/connectors/")
assert resp.status_code == 200
ids = [c["connector_id"] for c in resp.json()]
# With default settings, only echo is registered
assert "bascule" not in ids
assert "powershell" not in ids
assert "ansible" not in ids