flayer.integrations.wire

Module source

flayer.integrations.wire

Bounded strict JSON codecs validate representations without dispatch or authorization.

MAX_JSON_DEPTH module-attribute

MAX_JSON_DEPTH = 8

MAX_JSON_VALUES module-attribute

MAX_JSON_VALUES = 4096

ENVELOPE_FIELDS module-attribute

ENVELOPE_FIELDS = frozenset({'schema_version', 'envelope', 'identity', 'request_id', 'operation'})

ParseRequest

ParseRequest(payload: str, expected_identity: StackIdentity) -> IntegrationRequest

Validate untrusted read-only intent without invoking any operation or permission check.

Source code in installed/flayer/integrations/wire.py

def ParseRequest(payload: str, expected_identity: StackIdentity) -> IntegrationRequest:
    """Validate untrusted read-only intent without invoking any operation or permission check."""

    try:
        return _Request(_Decode(payload), expected_identity, "request")

    except ValueError:
        raise IntegrationError("Unable to parse a valid owned integration request") from None

ParseResult

ParseResult(payload: str, expected_request: IntegrationRequest) -> IntegrationResult

Bind an untrusted representation to its request without asserting authenticity.

Source code in installed/flayer/integrations/wire.py

def ParseResult(payload: str, expected_request: IntegrationRequest) -> IntegrationResult:
    """Bind an untrusted representation to its request without asserting authenticity."""

    try:
        if not isinstance(expected_request, IntegrationRequest):
            raise IntegrationError("Integration result requires an originating request")

        value = _Decode(payload)
        request = _Request(value, expected_request.identity, "result")

        if request != expected_request:
            raise IntegrationError("Integration result does not match the originating request")

        table = RequireTable(value, "integration result")

        if request.operation is IntegrationOperation.STATE:
            state = RequireTable(table["evidence"], "state evidence")
            ValidateFields(state, frozenset({"resource_count"}), frozenset(), "state evidence")

            return IntegrationResult(request, StateEvidence(cast(int, state["resource_count"])))

        if request.operation is IntegrationOperation.LIFECYCLE_REPORT:
            return IntegrationResult(request, _ParseLifecycle(table["evidence"]))

        return IntegrationResult(request, _ParseDiagnostic(table["evidence"]))

    except (ValueError, TypeError):
        raise IntegrationError("Unable to parse a valid correlated integration result") from None

SerializeRequest

SerializeRequest(request: IntegrationRequest) -> str

Serialize validated observation intent without resolving targets or credentials.

Source code in installed/flayer/integrations/wire.py

def SerializeRequest(request: IntegrationRequest) -> str:
    """Serialize validated observation intent without resolving targets or credentials."""

    if not isinstance(request, IntegrationRequest):
        raise IntegrationError("Integration serialization requires a validated request")

    return _Encode(_Metadata(request, "request"))

SerializeResult

SerializeResult(result: IntegrationResult) -> str

Serialize only explicitly allowed evidence, never source objects or raw adapter data.

Source code in installed/flayer/integrations/wire.py

def SerializeResult(result: IntegrationResult) -> str:
    """Serialize only explicitly allowed evidence, never source objects or raw adapter data."""

    if not isinstance(result, IntegrationResult):
        raise IntegrationError("Integration serialization requires a validated result")

    value = _Metadata(result.request, "result")
    evidence = result.evidence

    if isinstance(evidence, DiagnosticEvidence):
        value["evidence"] = {
            "status": evidence.Status.value,
            "checks": [{"status": check.status.value,
                        "metrics": {metric.name: metric.value for metric in check.metrics}}
                       for check in evidence.checks],
        }

    elif isinstance(evidence, StateEvidence):
        value["evidence"] = {"resource_count": evidence.resource_count}

    else:
        value["evidence"] = {
            "action": evidence.action, "status": evidence.status,
            "recovery_required": evidence.recovery_required,
            "resources": [{"status": resource.status, "count": resource.count}
                          for resource in evidence.resources],
        }

    return _Encode(value)