Skip to main content

hub_trigger

Verification of Hub-originated JOB_REQUEST_FROM_HUB triggers.

The Hub starts a v9 interactive DAG run on a pod with no modeller in the loop. There is no Access Manager authorisation and no modeller keypair in that flow, so the trigger authenticates itself instead: the Hub RSA-hybrid-encrypts the flow spec to the pod's public key, signs the ciphertext with a Hub signing key, and the pod verifies that signature against the public halves the Hub publishes as a JWKS.

The message body is plain JSON, camelCase throughout — clear routing fields plus a signedPayload blob:

{
  "keyId": ..., "datasetOwner": ..., "datasetName": ..., "projectId": ...,
  "taskId": ..., "taskRunMetadata": {...},
  "signedPayload": {"encryptedRequest": <b64>, "signature": <b64>}
}

encryptedRequest decrypts to zstd(msgpack({"flow_spec_raw": {...}})). Note the two layers differ deliberately: the outer body is plain JSON because the Hub serialises it, while the inner payload uses the SDK's own msgpack+zstd framing so a large flow spec does not become a large ciphertext. Only that inner payload is encrypted and signed; routing is read from the clear outer fields. That is deliberate — see accepted risk 6 in sdk/docs/adr/0003-hub-originated-job-triggers.md — and rests on hub→Kinesis→ Lambda being an IAM-trusted path.

The order of operations in HubTriggerVerifier.verify is load-bearing and must not be reordered: resolve keyId → verify the signature → only then decrypt with the pod's private key → then check freshness. Unauthenticated ciphertext is never fed to the pod's private key.

Module​

Global variables​

  • Base64Bytes - Bytes carried through JSON as base64. Decoding happens in validation, so every consumer downstream sees bytes and never has to remember to decode.

Classes​

HubTriggerError​

class HubTriggerError(*args, **kwargs):

Base class for JOB_REQUEST_FROM_HUB rejections.

Subclasses​

HubTriggerVerifier​

class HubTriggerVerifier(    jwk_client: PyJWKClient,    freshness_seconds: int,    unknown_key_id_ttl_seconds: int = 300,):

Authenticates JOB_REQUEST_FROM_HUB messages.

Holds the JWKS client (which owns key fetching, caching and rotation) and the negative cache for unknown kids. Build one per pod and reuse it, so the caches are actually shared across triggers.

Arguments

  • jwk_client: Client for the Hub's JWKS endpoint. Used only for fetch/cache/rotation — the returned PyJWK's raw key is handed to _RSAEncryption; jwt.decode is never called, because these signatures are detached RSA-PSS over ciphertext rather than JWS.
  • freshness_seconds: How old a trigger may be before it is rejected as stale.
  • unknown_key_id_ttl_seconds: How long an unresolvable kid is remembered so it cannot force repeated JWKS refetches.

Static methods​

from_config​

def from_config() ‑> HubTriggerVerifier:

Build a verifier from config.settings.

The JWKS URI and freshness window are read once here, so a later config.refresh_configuration() does not affect a running pod; rebuild the verifier (restart the pod) to pick up such a change.

Methods​

verify​

def verify(    self, message: _BitfountMessage, pod_private_key: RSAPrivateKey,) ‑> VerifiedHubJobRequest:

Authenticate a JOB_REQUEST_FROM_HUB and return its contents.

The step order here is load-bearing: the signature is checked before the ciphertext ever reaches the pod's private key, and freshness is checked last so a stale-but-authentic trigger is distinguishable from a forged one.

Arguments

  • message: The incoming JOB_REQUEST_FROM_HUB message.
  • pod_private_key: The pod's RSA private key, which the payload was encrypted to.

Returns The verified request: clear routing fields, decrypted flow spec, and the Hub key to encrypt replies back to.

Raises

  • MalformedHubTrigger: If the body cannot be parsed at all.
  • InvalidHubSignature: If the keyId does not resolve, the signature does not verify, or the payload does not decrypt.
  • StaleHubTrigger: If the envelope timestamp is outside the freshness window.

InvalidHubSignature​

class InvalidHubSignature(*args, **kwargs):

The trigger's signature did not verify against the Hub's signing keys.

Covers an unresolvable keyId, a signature made with the wrong key, tampered ciphertext, and a payload the pod's private key cannot decrypt — all of which mean the same thing to the caller: this did not come intact from the Hub. Maps to Reason.INVALID_SIGNATURE.

Ancestors​

MalformedHubTrigger​

class MalformedHubTrigger(*args, **kwargs):

The message body is not a well-formed hub trigger payload.

Distinct from InvalidHubSignature: nothing was verifiable in the first place, so there is no signature verdict to report.

Ancestors​

StaleHubTrigger​

class StaleHubTrigger(*args, **kwargs):

The trigger arrived outside the accepted freshness window.

Maps to Reason.STALE_REQUEST. This is staleness control for backlogged or redelivered triggers (e.g. after a pod outage), not replay defence — the envelope timestamp this is measured against is not integrity-protected.

Ancestors​

VerifiedHubJobRequest​

class VerifiedHubJobRequest(**data: Any):

A JOB_REQUEST_FROM_HUB that verified, decrypted and is fresh.

Attributes

  • payload: The clear routing fields the run is dispatched on.
  • flow_spec_raw: The v9 FlowSpec, as decrypted from encrypted_request. This is the only integrity-protected part of the message.
  • hub_public_key: The Hub key the signature verified against. Replies are encrypted to this same key — one Hub keypair serves both signing and reply-decryption, so a rotation makes in-flight replies undecryptable (accepted risk 4).

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static model_config
  • static payload : bitfount.federated.hub_trigger._JobRequestFromHubPayload