eee8740ce8
Bascule: session-based connector using AC as credential. Transport stubbed — Shellstream integration in future sprint. PowerShell: session-based connector using Kerberos credentials from CredentialResolver. PSRP transport stubbed — pypsrp integration in future sprint. Signed-off-by: Tyler King <tking@guildhouse.dev>
87 lines
3 KiB
Python
87 lines
3 KiB
Python
# Copyright 2026 Guildhouse Dev
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
"""Bascule connector — governed shell sessions via Shellstream.
|
|
|
|
The Bascule connector uses the AC as the credential. Bascule (the
|
|
governed-shell runtime) validates ACs natively — no separate secrets
|
|
backend is needed. The transport establishes a Shellstream session
|
|
to the target endpoint via a Bascule proxy.
|
|
|
|
Real integration:
|
|
The ``BasculeTransport`` will use the Shellstream protocol library
|
|
(``substrate/shellstream/``) to establish a 3-way attested handshake
|
|
with the target. The SAT (Shell Attestation Token) embedded in the
|
|
Shellstream frame header is derived from the AC. See
|
|
``SS-SPEC-0001 — Shellstream Protocol.md`` for the handshake flow.
|
|
|
|
Library: ``shellstream-py`` (future) or gRPC to Bascule proxy
|
|
Protocol: Shellstream over TCP / QUIC
|
|
Auth: AC passthrough — the AC IS the credential
|
|
|
|
Stubbed in this sprint — transport returns placeholder results.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Optional
|
|
|
|
from gsap_broker.connectors.session import SessionConnector, SessionTransport
|
|
from gsap_broker.credentials.resolver import Credential, CredentialResolver
|
|
|
|
|
|
class BasculeTransport(SessionTransport):
|
|
"""Shellstream transport to a Bascule-governed endpoint.
|
|
|
|
Stubbed — actual Shellstream integration in a future sprint.
|
|
"""
|
|
|
|
transport_id = "bascule"
|
|
|
|
def __init__(self) -> None:
|
|
self._target = ""
|
|
self._connected = False
|
|
|
|
async def connect(self, target: str, credential: Credential) -> None:
|
|
# TODO: establish Shellstream connection to target via Bascule
|
|
# proxy. The credential IS the AC (BasculeCredential).
|
|
# Handshake: ATTEST-INIT → ATTEST-VERIFY → ATTEST-CONFIRM.
|
|
self._target = target
|
|
self._connected = True
|
|
|
|
async def execute(self, command: str, params: Optional[dict[str, Any]] = None) -> dict[str, Any]:
|
|
# TODO: send command via Shellstream, capture output.
|
|
# Shellstream frames carry the SAT in the header for
|
|
# per-frame attestation.
|
|
if not self._connected:
|
|
raise RuntimeError("Not connected")
|
|
return {
|
|
"stub": True,
|
|
"transport": self.transport_id,
|
|
"target": self._target,
|
|
"command": command,
|
|
"params": params or {},
|
|
}
|
|
|
|
async def disconnect(self) -> None:
|
|
self._connected = False
|
|
|
|
async def is_alive(self) -> bool:
|
|
return self._connected
|
|
|
|
|
|
class BasculeConnector(SessionConnector):
|
|
"""Governed shell connector using Bascule/Shellstream."""
|
|
|
|
connector_id = "bascule"
|
|
corpus_entry_cid = "sha256:bascule-connector-v1"
|
|
credential_type = "bascule_ac"
|
|
transport_class = BasculeTransport
|
|
capability_mask = 0x7 # READ | PROPOSE | MUTATE
|
|
declared_endpoints = ["shellstream://*"]
|
|
accord_template = "governed-shell"
|
|
gsap_required = True
|
|
chronicle_enabled = True
|
|
|
|
def __init__(self, credential_resolver: CredentialResolver):
|
|
super().__init__(credential_resolver)
|