e744336385
C-6: ConnectorRuntime enforces capability_mask per operation.
READ-only ACs cannot invoke MUTATE operations (wipe, lock, retire).
C-7: AC validated against database (exists, active, not expired)
before connector invocation.
C-9: Delegated AC capability bounded by delegator's capability.
C-10: Command counter uses atomic SQL increment with limit check.
M-23: expire_stale() uses same atomic SQL pattern.
H-1: Sensitive credential fields hidden from repr/logs via repr=False.
H-2: Stub backend requires ALLOW_STUB_CREDENTIALS=true to activate.
H-3: Kerberos backend raises CredentialResolutionError instead of
returning stub ticket.
H-4: Chronicle INTENT emitted before execution, RESULT after.
H-5: device_id validated as UUID before Graph API URL interpolation.
H-8: ConnectorRuntime enforces governance for all connector invocations.
Signed-off-by: Tyler King <tking@guildhouse.dev>
81 lines
2.5 KiB
Python
81 lines
2.5 KiB
Python
"""ConnectorPlugin ABC per GCAP-SPEC-CONNECTOR-DESCRIPTOR-0001 section 4."""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from abc import ABC, abstractmethod
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
# Fix C-6: capability constants for operation-level enforcement
|
|
CAP_READ = 0x1
|
|
CAP_PROPOSE = 0x2
|
|
CAP_MUTATE = 0x4
|
|
CAP_ADMIN = 0x8
|
|
|
|
|
|
@dataclass
|
|
class ConnectorContext:
|
|
chronicle_session_id: str = ""
|
|
gsap_context_id: str = ""
|
|
credentials: dict[str, Any] = field(default_factory=dict)
|
|
pipeline_run_id: str = ""
|
|
dag_id: str = ""
|
|
# Fix C-6: capability_mask from the AC, enforced by ConnectorRuntime
|
|
capability_mask: int = 0
|
|
principal_did: str = ""
|
|
|
|
|
|
@dataclass
|
|
class ConnectorResult:
|
|
success: bool = False
|
|
data: Any = None
|
|
error: str | None = None
|
|
lineage_cid: str = ""
|
|
metadata: dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
class ConnectorPlugin(ABC):
|
|
"""Abstract base for governed connectors."""
|
|
|
|
connector_id: str = ""
|
|
corpus_entry_cid: str = ""
|
|
capability_mask: int = 0
|
|
declared_endpoints: list[str] = []
|
|
accord_template: str = ""
|
|
gsap_required: bool = True
|
|
chronicle_enabled: bool = True
|
|
# Fix C-6: per-operation capability requirements
|
|
operation_capabilities: dict[str, int] = {}
|
|
|
|
def capability_for_operation(self, operation: str) -> int:
|
|
"""Return the capability mask required for an operation.
|
|
Defaults to CAP_READ if not explicitly mapped."""
|
|
return self.operation_capabilities.get(operation, CAP_READ)
|
|
|
|
@abstractmethod
|
|
async def invoke(
|
|
self, operation: str, parameters: dict[str, Any], context: ConnectorContext
|
|
) -> ConnectorResult:
|
|
...
|
|
|
|
@abstractmethod
|
|
def health_check(self) -> bool:
|
|
...
|
|
|
|
def compute_parameters_cid(self, parameters: dict[str, Any]) -> str:
|
|
canonical = json.dumps(parameters, sort_keys=True, separators=(",", ":"))
|
|
return "sha256:" + hashlib.sha256(canonical.encode()).hexdigest()
|
|
|
|
def descriptor(self) -> dict[str, Any]:
|
|
return {
|
|
"@context": "https://schema.gsap.dev/connector/v1",
|
|
"@type": "ConnectorDescriptor",
|
|
"connector_id": self.connector_id,
|
|
"corpus_entry_cid": self.corpus_entry_cid,
|
|
"capability_mask": self.capability_mask,
|
|
"declared_endpoints": self.declared_endpoints,
|
|
"accord_template": self.accord_template,
|
|
"gsap_required": self.gsap_required,
|
|
"chronicle_enabled": self.chronicle_enabled,
|
|
}
|