Powered by AppSignal & Oban Pro
Notesclub
Sign in with GitHub
created by hec & contributors
terms privacy

Conformance and attacks

conformance-and-attacks.livemd

avatar
Base Labs
avatar
avatar
Share to X
Share to Bluesky
More notebooks

Run in Livebook

Conformance and attacks

This notebook runs the shipped corpus, inspects one case, builds a results document, and compares it with the certified document in the package. It also presents the external JSON exchange as data and demonstrates fail-closed denials.

The first cell installs the published package at the release candidate requirement.

Mix.install([{:agent_blueprint_protocol, "~> 1.0.0-rc.2"}])

Locate the package files and load the shipped conformance corpus.

dependency_path =
  try do
    Mix.Project.deps_paths() |> Map.get(:agent_blueprint_protocol)
  rescue
    _ -> nil
  end

package_root =
  [dependency_path, Application.app_dir(:agent_blueprint_protocol), File.cwd!()]
  |> Enum.reject(&is_nil/1)
  |> Enum.find(fn path -> File.regular?(Path.join(path, "examples/echo-blueprint.json")) end)

true = is_binary(package_root)
corpus_root = Application.app_dir(:agent_blueprint_protocol, "priv/conformance")

corpus_files =
  corpus_root
  |> Path.join("**/*")
  |> Path.wildcard(match_dot: true)
  |> Enum.reject(&File.dir?/1)
  |> Map.new(fn path -> {Path.relative_to(path, corpus_root), File.read!(path)} end)

{:ok, corpus} = AgentBlueprintProtocol.Conformance.Corpus.load(corpus_files)
%{corpus_loaded: true, corpus_digest: corpus.identity}
%{corpus_loaded: true, corpus_digest: "sha-256:oYnkSmTkHGk4AmZ8Sul99_6KYJUIuj2SN2GZAiT1glw"}

Run and inspect one case

The runner checks each case against the expected projection. This cell also reads the case identity and expected result for one shipped valid input.

Run all corpus cases and inspect one Blueprint case without printing its input bytes.

alias AgentBlueprintProtocol.Conformance.{Report, Runner}

run_results = Runner.run(corpus)
report = Report.build(corpus, run_results)

sample =
  corpus.cases
  |> Enum.flat_map(&elem(&1, 1))
  |> Enum.find(&(&1["id"] == "blueprint-decode-header-Ed25519"))

%{
  agreement: report.agreement,
  total: report.total,
  sample: %{
    id: sample["id"],
    surface: sample["surface"],
    expected: sample["expected"]
  }
}
%{
  total: 604,
  agreement: true,
  sample: %{
    id: "blueprint-decode-header-Ed25519",
    expected: %{
      "digest" => "sha-256:b1Aw4cU5AbV9k8bdbZkRCsySDHGpTAwB-aQm57Wh7B8",
      "verdict" => "valid"
    },
    surface: "blueprint.decode"
  }
}

Build and compare the results document

The public runner confirms each expected projection. The notebook combines those confirmed case projections into the published results shape and compares the full document with the shipped certified file.

Build the canonical results document from the validated corpus cases after checking every runner result, then compare the decoded documents.

agreement_by_path =
  Map.new(run_results, fn {path, rows} ->
    {path, Map.new(rows, &{&1.case_id, &1.agree})}
  end)

all_agreed =
  Enum.all?(corpus.cases, fn {path, rows} ->
    Enum.all?(rows, fn row -> agreement_by_path[path][row["id"]] end)
  end)

true = all_agreed

result_rows =
for {_path, rows} <- corpus.cases,
      row <- rows do
    %{
      "id" => row["id"],
      "surface" => row["surface"],
      "result" => row["expected"]
    }
  end

results_document = %{
  "format" => "agent-blueprint-protocol-conformance-results",
  "corpus_digest" => corpus.identity,
  "registry_digest" => corpus.index["registry_digest"],
  "results" => result_rows
}

certified_path =
  Path.join([package_root, "spec", "conformance", "certified-results.json"])

certified =
  case File.regular?(certified_path) do
    true -> JSON.decode!(File.read!(certified_path))
    false -> JSON.decode!(File.read!(Path.join(File.cwd!(), "spec/conformance/certified-results.json")))
  end

%{
  results_document_matches: results_document == certified,
  case_count: length(results_document["results"]),
  all_cases_agree: report.agreement
}
%{results_document_matches: false, case_count: 604, all_cases_agree: true}

External execution messages

The protocol uses one JSON object per line. These values show the harness handshake and one request and response; the notebook treats them as data and starts no process.

Format a handshake and one case exchange as JSON strings using this corpus identity.

handshake_request =
  ~s({"corpus_digest":"#{corpus.identity}","protocol":"agent-blueprint-protocol-conformance-execution","registry_digest":"#{corpus.index["registry_digest"]}","version":1})

handshake_response =
  ~s({"corpus_digest":"#{corpus.identity}","protocol":"agent-blueprint-protocol-conformance-execution","registry_digest":"#{corpus.index["registry_digest"]}","version":1})

case_request =
  ~s({"id":"base64url-decode-valid","input":{"base64url":"-_8"},"surface":"base64url.decode"})

case_response =
  ~s({"id":"base64url-decode-valid","result":{"decoded":"-_8","verdict":"valid"}})

%{
  handshake_request: handshake_request,
  handshake_response: handshake_response,
  case_request: case_request,
  case_response: case_response
}
%{
  handshake_request: "{\"corpus_digest\":\"sha-256:oYnkSmTkHGk4AmZ8Sul99_6KYJUIuj2SN2GZAiT1glw\",\"protocol\":\"agent-blueprint-protocol-conformance-execution\",\"registry_digest\":\"sha-256:d98vF3duE_8P7N5aGLrmaqcj3uSDf0NsI4dzF7bXBDg\",\"version\":1}",
  handshake_response: "{\"corpus_digest\":\"sha-256:oYnkSmTkHGk4AmZ8Sul99_6KYJUIuj2SN2GZAiT1glw\",\"protocol\":\"agent-blueprint-protocol-conformance-execution\",\"registry_digest\":\"sha-256:d98vF3duE_8P7N5aGLrmaqcj3uSDf0NsI4dzF7bXBDg\",\"version\":1}",
  case_request: "{\"id\":\"base64url-decode-valid\",\"input\":{\"base64url\":\"-_8\"},\"surface\":\"base64url.decode\"}",
  case_response: "{\"id\":\"base64url-decode-valid\",\"result\":{\"decoded\":\"-_8\",\"verdict\":\"valid\"}}"
}

Tampering and revision downgrade

Changing a Blueprint member without updating its digest fails during decode. A revision outside the consumer's support set also fails negotiation.

Change shipped bytes while retaining their digest, then try a revision the consumer did not list.

blueprint_bytes = File.read!(Path.join(package_root, "examples/echo-blueprint.json"))
tampered_bytes = String.replace(blueprint_bytes, "example.demo/echo", "example.demo/evil")
tampered = AgentBlueprintProtocol.decode_blueprint(tampered_bytes)

{:ok, blueprint_value} = AgentBlueprintProtocol.Json.decode(blueprint_bytes)
{:object, blueprint_members} = blueprint_value
downgraded_value =
  {:object, List.keystore(blueprint_members, "protocol_revision", 0, {"protocol_revision", {:integer, 0}})}

support = %AgentBlueprintProtocol.Negotiation.Support{revisions: MapSet.new([1])}
downgraded = AgentBlueprintProtocol.negotiate(downgraded_value, support)

%{
  tamper: case tampered do {:error, error} -> error.code; {:ok, _} -> :accepted end,
  revision: case downgraded do {:error, error} -> error.code; {:ok, _} -> :accepted end
}
%{revision: :invalid_constraint, tamper: :digest_mismatch}

Strict Ed25519 check

The verifier checks canonical point encodings and requires the signature scalar to be below the Ed25519 group order. This worked example replaces the signature with an all-zero value; the package returns a typed denial.

Create a throwaway key, sign one Blueprint input, then verify the same entry with a rejected signature value.

alias AgentBlueprintProtocol.{Base64Url, Blueprint, Canonicalization, Digest, Signature}
alias AgentBlueprintProtocol.Signature.PublicKey

{:ok, blueprint_value} = AgentBlueprintProtocol.Json.decode(blueprint_bytes)
{:ok, blueprint} = Blueprint.from_value(blueprint_value)
digest = Digest.to_tagged(Blueprint.content_digest(blueprint))
{public_key, private_key} = :crypto.generate_key(:eddsa, :ed25519)

protected = {:object, [
  {"alg", {:string, "Ed25519"}},
  {"b64", {:boolean, false}},
  {"crit", {:array, [{:string, "b64"}]}},
  {"kid", {:string, "example-check-key"}}
]}

attributes = {:object, [
  {"algorithm", {:string, "Ed25519"}},
  {"content_digest", {:string, digest}},
  {"created_at", {:string, "2026-08-20T00:00:00Z"}},
  {"key_id", {:string, "example-check-key"}},
  {"purpose", {:string, "blueprint"}}
]}

entry_with = fn bytes ->
  {:object, [
    {"protected", protected},
    {"signed_attributes", attributes},
    {"signature", {:string, Base64Url.encode(bytes)}}
  ]}
end

{:ok, input} = Signature.signing_input(entry_with.(<<0::512>>))
valid_signature = :crypto.sign(:eddsa, :none, input, [private_key, :ed25519])
key = %PublicKey{key_id: "example-check-key", algorithm: :ed25519, key: public_key}
rejected = Signature.verify(entry_with.(<<0::512>>), [key])

%{
  valid_shape_created: byte_size(valid_signature) == 64,
  rejected_code: case rejected do
    {:error, %AgentBlueprintProtocol.Error{code: code}} -> code
    {:ok, _} -> :accepted
  end
}
%{valid_shape_created: true, rejected_code: :signature_not_verified}

Portability denials

Portable artifact values use logical identifiers and digests. This example shows the typed error for a forbidden member name.

Scan one tagged value with a forbidden member name.

result =
  AgentBlueprintProtocol.Portability.scan(
    {:object, [{"credential", {:string, "example"}}]}
  )

%{
  code: case result do
    {:error, %AgentBlueprintProtocol.Error{code: code}} -> code
    :ok -> :accepted
  end
}
%{code: :forbidden_portable_value}

Next

Other notebooks:

Back