use buffa::Message as _;
use polyc_proto::proto::polychrome::approval::v1::{ApprovalResponseRequest, AssertedApproval};
use crate::{Signer, verify};
pub const APPROVAL_ASSERTION_DOMAIN_PREFIX: &[u8] = b"polychrome.approval-assertion.v1\0";
#[must_use]
pub fn canonical_bytes(request: &ApprovalResponseRequest) -> Vec<u8> {
let mut canonical = request.clone();
if let Some(assertion) = canonical.asserted_approval.as_option_mut() {
assertion.signature_hex.clear();
}
let encoded = canonical.encode_to_vec();
let mut bytes = Vec::with_capacity(APPROVAL_ASSERTION_DOMAIN_PREFIX.len() + encoded.len());
bytes.extend_from_slice(APPROVAL_ASSERTION_DOMAIN_PREFIX);
bytes.extend_from_slice(&encoded);
bytes
}
#[must_use]
pub fn sign_approval_assertion(signer: &Signer, request: &ApprovalResponseRequest) -> String {
crate::hex::lower(&signer.sign(&canonical_bytes(request)))
}
pub fn attach_approval_assertion(
signer: &Signer,
request: &mut ApprovalResponseRequest,
assertion: AssertedApproval,
) {
request.asserted_approval = buffa::MessageField::some(assertion);
let signature = sign_approval_assertion(signer, request);
if let Some(attached) = request.asserted_approval.as_option_mut() {
attached.signature_hex = signature;
}
}
#[must_use]
pub fn verify_approval_assertion<'a, K>(
candidate_public_keys: K,
request: &ApprovalResponseRequest,
) -> bool
where
K: IntoIterator<Item = &'a [u8]>,
{
let Some(assertion) = request.asserted_approval.as_option() else {
return false;
};
let Some(signature) = crate::hex::decode(&assertion.signature_hex) else {
return false;
};
let bytes = canonical_bytes(request);
let mut verified = false;
for public_key in candidate_public_keys {
if verify(public_key, &bytes, &signature) {
verified = true;
}
}
verified
}
#[cfg(test)]
mod tests {
#![allow(clippy::pedantic, clippy::nursery, missing_docs)]
use polyc_proto::proto::polychrome::approval::v1::{
Approve, Deny, approval_response_request::Decision,
};
use polyc_proto::proto::polychrome::persona::v1::ExternalIdentity;
use super::*;
fn sample_responder() -> ExternalIdentity {
ExternalIdentity {
provider: "chat".to_owned(),
scope: "team-1".to_owned(),
external_id: "U123".to_owned(),
display_name: "Ada".to_owned(),
..Default::default()
}
}
fn sample_assertion() -> AssertedApproval {
AssertedApproval {
edge_id: "chat".to_owned(),
responder: buffa::MessageField::some(sample_responder()),
..Default::default()
}
}
fn sample_request() -> ApprovalResponseRequest {
ApprovalResponseRequest {
request_id: "req-1".to_owned(),
conversation_id: "chat:team-1:general".to_owned(),
resolve_token: "token-1".to_owned(),
decision: Some(Decision::Approve(Box::new(Approve {
reason: "looks right".to_owned(),
..Default::default()
}))),
..Default::default()
}
}
fn signed_request(signer: &Signer) -> ApprovalResponseRequest {
let mut request = sample_request();
attach_approval_assertion(signer, &mut request, sample_assertion());
request
}
fn keys(public_key: &[u8]) -> [&[u8]; 1] {
[public_key]
}
#[test]
fn round_trips() {
let signer = Signer::from_seed(41);
let pk = signer.public_key_bytes();
let request = signed_request(&signer);
assert!(
!request
.asserted_approval
.as_option()
.unwrap()
.signature_hex
.is_empty()
);
assert!(verify_approval_assertion(keys(&pk), &request));
}
#[test]
fn tampered_responder_fails() {
let signer = Signer::from_seed(41);
let pk = signer.public_key_bytes();
let mut request = signed_request(&signer);
request.asserted_approval.as_option_mut().unwrap().responder =
buffa::MessageField::some(ExternalIdentity {
external_id: "U999".to_owned(),
..sample_responder()
});
assert!(
!verify_approval_assertion(keys(&pk), &request),
"swapping the asserted responder must invalidate the signature (INV-A3)"
);
}
#[test]
fn tampered_decision_fails() {
let signer = Signer::from_seed(41);
let pk = signer.public_key_bytes();
let mut request = signed_request(&signer);
request.decision = Some(Decision::Deny(Box::new(Deny {
reason: "looks right".to_owned(),
..Default::default()
})));
assert!(
!verify_approval_assertion(keys(&pk), &request),
"swapping approve for deny must invalidate the signature (INV-A5)"
);
}
#[test]
fn tampered_modified_args_fails() {
let signer = Signer::from_seed(41);
let pk = signer.public_key_bytes();
let mut request = signed_request(&signer);
request.decision = Some(Decision::Approve(Box::new(Approve {
modified_args_json: r#"{"path":"/"}"#.to_owned(),
reason: "looks right".to_owned(),
..Default::default()
})));
assert!(!verify_approval_assertion(keys(&pk), &request));
}
#[test]
fn tampered_request_id_fails() {
let signer = Signer::from_seed(41);
let pk = signer.public_key_bytes();
let mut request = signed_request(&signer);
request.request_id = "req-2".to_owned();
assert!(
!verify_approval_assertion(keys(&pk), &request),
"an assertion is bound to one request_id (INV-A5)"
);
}
#[test]
fn tampered_conversation_id_fails() {
let signer = Signer::from_seed(41);
let pk = signer.public_key_bytes();
let mut request = signed_request(&signer);
request.conversation_id = "chat:team-1:secrets".to_owned();
assert!(
!verify_approval_assertion(keys(&pk), &request),
"an assertion is bound to one conversation_id (INV-A5)"
);
}
#[test]
fn tampered_resolve_token_fails() {
let signer = Signer::from_seed(41);
let pk = signer.public_key_bytes();
let mut request = signed_request(&signer);
request.resolve_token = "token-2".to_owned();
assert!(!verify_approval_assertion(keys(&pk), &request));
}
#[test]
fn tampered_edge_id_fails() {
let signer = Signer::from_seed(41);
let pk = signer.public_key_bytes();
let mut request = signed_request(&signer);
request.asserted_approval.as_option_mut().unwrap().edge_id = "messaging".to_owned();
assert!(!verify_approval_assertion(keys(&pk), &request));
}
#[test]
fn wrong_key_fails() {
let signer = Signer::from_seed(41);
let other = Signer::from_seed(42);
let request = signed_request(&signer);
assert!(!verify_approval_assertion(
keys(&other.public_key_bytes()),
&request
));
}
#[test]
fn verifies_against_any_candidate_key() {
let signer = Signer::from_seed(41);
let retiring = Signer::from_seed(42);
let retiring_pk = retiring.public_key_bytes();
let active_pk = signer.public_key_bytes();
let request = signed_request(&signer);
let candidates: [&[u8]; 2] = [&retiring_pk, &active_pk];
assert!(verify_approval_assertion(candidates, &request));
}
#[test]
fn no_candidate_keys_fails() {
let signer = Signer::from_seed(41);
let request = signed_request(&signer);
let candidates: [&[u8]; 0] = [];
assert!(!verify_approval_assertion(candidates, &request));
}
#[test]
fn absent_assertion_fails() {
let signer = Signer::from_seed(41);
let pk = signer.public_key_bytes();
let request = sample_request();
assert!(!verify_approval_assertion(keys(&pk), &request));
}
#[test]
fn unsigned_assertion_fails() {
let signer = Signer::from_seed(41);
let pk = signer.public_key_bytes();
let mut request = sample_request();
request.asserted_approval = buffa::MessageField::some(sample_assertion());
assert!(!verify_approval_assertion(keys(&pk), &request));
}
#[test]
fn garbage_signature_hex_returns_false_not_panic() {
let signer = Signer::from_seed(41);
let pk = signer.public_key_bytes();
let mut request = signed_request(&signer);
for garbage in ["not-hex!", "abc", "", "zz", &"ab".repeat(4096)] {
request
.asserted_approval
.as_option_mut()
.unwrap()
.signature_hex = garbage.to_owned();
assert!(
!verify_approval_assertion(keys(&pk), &request),
"malformed signature_hex {garbage:?} must return false, not panic"
);
}
}
#[test]
fn a_stale_signature_does_not_change_the_minted_one() {
let signer = Signer::from_seed(41);
let pk = signer.public_key_bytes();
let mut clean = sample_request();
attach_approval_assertion(&signer, &mut clean, sample_assertion());
let mut stale = sample_request();
attach_approval_assertion(
&signer,
&mut stale,
AssertedApproval {
signature_hex: "deadbeef".to_owned(),
..sample_assertion()
},
);
assert_eq!(
clean.asserted_approval.as_option().unwrap().signature_hex,
stale.asserted_approval.as_option().unwrap().signature_hex
);
assert!(verify_approval_assertion(keys(&pk), &stale));
}
#[test]
fn undomained_signature_is_rejected() {
let signer = Signer::from_seed(41);
let pk = signer.public_key_bytes();
let mut request = sample_request();
request.asserted_approval = buffa::MessageField::some(sample_assertion());
let raw = request.encode_to_vec();
request
.asserted_approval
.as_option_mut()
.unwrap()
.signature_hex = crate::hex::lower(&signer.sign(&raw));
assert!(!verify_approval_assertion(keys(&pk), &request));
}
#[test]
fn edge_identity_signature_never_verifies_as_an_approval_assertion() {
let signer = Signer::from_seed(41);
let pk = signer.public_key_bytes();
let mut request = sample_request();
request.asserted_approval = buffa::MessageField::some(sample_assertion());
let encoded = request.encode_to_vec();
let mut cross_domain = Vec::new();
cross_domain.extend_from_slice(crate::edge_identity::EDGE_IDENTITY_DOMAIN_PREFIX);
cross_domain.extend_from_slice(&encoded);
request
.asserted_approval
.as_option_mut()
.unwrap()
.signature_hex = crate::hex::lower(&signer.sign(&cross_domain));
assert!(
!verify_approval_assertion(keys(&pk), &request),
"a signature minted under the edge-identity prefix must not verify as an approval \
assertion (INV-A4)"
);
let mut same_domain = Vec::new();
same_domain.extend_from_slice(APPROVAL_ASSERTION_DOMAIN_PREFIX);
same_domain.extend_from_slice(&encoded);
request
.asserted_approval
.as_option_mut()
.unwrap()
.signature_hex = crate::hex::lower(&signer.sign(&same_domain));
assert!(verify_approval_assertion(keys(&pk), &request));
}
#[test]
fn approval_assertion_signature_never_verifies_as_an_edge_identity_envelope() {
use polyc_proto::proto::polychrome::agent::v1::AssertedAttribution;
let signer = Signer::from_seed(41);
let pk = signer.public_key_bytes();
let mut envelope = AssertedAttribution {
edge_id: "chat".to_owned(),
conversation_id: "chat:team-1:general".to_owned(),
nonce: "nonce-1".to_owned(),
issued_unix_ms: 1_700_000_000_000,
caller: buffa::MessageField::some(sample_responder()),
exec_id: "exec-1".to_owned(),
content_hash: crate::edge_identity::content_hash_hex(b"messages-bytes"),
..Default::default()
};
let encoded = envelope.encode_to_vec();
let mut cross_domain = Vec::new();
cross_domain.extend_from_slice(APPROVAL_ASSERTION_DOMAIN_PREFIX);
cross_domain.extend_from_slice(&encoded);
envelope.signature_hex = crate::hex::lower(&signer.sign(&cross_domain));
assert!(
!crate::edge_identity::verify_edge_assertion(&pk, &envelope),
"a signature minted under the approval-assertion prefix must not verify as an \
edge-identity envelope (INV-A4)"
);
crate::edge_identity::sign_edge_assertion_into(&signer, &mut envelope);
assert!(crate::edge_identity::verify_edge_assertion(&pk, &envelope));
}
#[test]
fn the_two_domain_prefixes_are_distinct_and_nul_terminated() {
assert_ne!(
APPROVAL_ASSERTION_DOMAIN_PREFIX,
crate::edge_identity::EDGE_IDENTITY_DOMAIN_PREFIX
);
assert_eq!(APPROVAL_ASSERTION_DOMAIN_PREFIX.last(), Some(&0u8));
}
}
#[cfg(test)]
mod conformance {
#![allow(clippy::pedantic, clippy::nursery, missing_docs)]
use buffa::Message as _;
use polyc_proto::proto::polychrome::approval::v1::{
Approve, Deny, approval_response_request::Decision,
};
use polyc_proto::proto::polychrome::persona::v1::ExternalIdentity;
use serde_json::Value;
use super::*;
const VECTORS: &str = polyc_conformance_vectors::APPROVAL_ASSERTION;
fn vectors() -> Value {
serde_json::from_str(VECTORS).expect("the conformance vector file is valid JSON")
}
fn hex_of(v: &Value, path: &[&str]) -> Vec<u8> {
let mut node = v;
for key in path {
node = &node[*key];
}
crate::hex::decode(node.as_str().expect("a hex string")).expect("valid hex")
}
fn vector_request(v: &Value) -> ApprovalResponseRequest {
let r = &v["vector"]["request"];
let approve = &r["decision"]["Approve"];
let responder = &r["asserted_approval"]["responder"];
ApprovalResponseRequest {
request_id: r["request_id"].as_str().unwrap().to_owned(),
conversation_id: r["conversation_id"].as_str().unwrap().to_owned(),
resolve_token: r["resolve_token"].as_str().unwrap().to_owned(),
decision: Some(Decision::Approve(Box::new(Approve {
modified_args_json: approve["modified_args_json"].as_str().unwrap().to_owned(),
injected_context: approve["injected_context"].as_str().unwrap().to_owned(),
reason: approve["reason"].as_str().unwrap().to_owned(),
approved_for_session: approve["approved_for_session"].as_bool().unwrap(),
..Default::default()
}))),
asserted_approval: buffa::MessageField::some(AssertedApproval {
edge_id: r["asserted_approval"]["edge_id"]
.as_str()
.unwrap()
.to_owned(),
responder: buffa::MessageField::some(ExternalIdentity {
provider: responder["provider"].as_str().unwrap().to_owned(),
scope: responder["scope"].as_str().unwrap().to_owned(),
external_id: responder["external_id"].as_str().unwrap().to_owned(),
display_name: responder["display_name"].as_str().unwrap().to_owned(),
time_zone: responder["time_zone"].as_str().unwrap().to_owned(),
..Default::default()
}),
..Default::default()
}),
..Default::default()
}
}
#[test]
fn the_pinned_domain_prefix_is_the_one_this_crate_uses() {
let v = vectors();
assert_eq!(
v["domain_separation"]["approval_assertion_prefix_ascii"]
.as_str()
.unwrap()
.as_bytes(),
APPROVAL_ASSERTION_DOMAIN_PREFIX
);
assert_eq!(
hex_of(&v, &["domain_separation", "approval_assertion_prefix_hex"]),
APPROVAL_ASSERTION_DOMAIN_PREFIX
);
assert_eq!(
v["domain_separation"]["edge_identity_prefix_ascii"]
.as_str()
.unwrap()
.as_bytes(),
crate::edge_identity::EDGE_IDENTITY_DOMAIN_PREFIX
);
assert_eq!(
v["signature_scheme"]["namespace_ascii"]
.as_str()
.unwrap()
.as_bytes(),
crate::NAMESPACE
);
}
#[test]
fn the_known_good_vector_reproduces_its_bytes_and_signature() {
let v = vectors();
let request = vector_request(&v);
assert_eq!(
request.encode_to_vec(),
hex_of(&v, &["vector", "expected_request_encoding_hex"]),
"the buffa encoding of the vector's request must match the pinned bytes"
);
let canonical = canonical_bytes(&request);
assert_eq!(
canonical,
hex_of(&v, &["vector", "expected_canonical_bytes_hex"]),
"canonical bytes must be the domain prefix followed by that encoding"
);
let mut preimage = vec![u8::try_from(crate::NAMESPACE.len()).unwrap()];
preimage.extend_from_slice(crate::NAMESPACE);
preimage.extend_from_slice(&canonical);
assert_eq!(
preimage,
hex_of(&v, &["vector", "expected_ed25519_message_hex"])
);
let private_key = hex_of(&v, &["signer", "ed25519_private_key_hex"]);
let signer = Signer::from_key_bytes(&private_key).expect("valid key material");
let public_key = signer.public_key_bytes();
assert_eq!(
public_key,
hex_of(&v, &["signer", "ed25519_public_key_hex"]),
"the pinned public key must be the one the pinned private key derives"
);
let expected_signature = v["vector"]["expected_signature_hex"].as_str().unwrap();
assert_eq!(
sign_approval_assertion(&signer, &request),
expected_signature,
"signing the vector's request must reproduce the pinned signature"
);
let mut signed = request;
signed
.asserted_approval
.as_option_mut()
.unwrap()
.signature_hex = expected_signature.to_owned();
assert!(
verify_approval_assertion([public_key.as_slice()], &signed),
"the known-good vector must verify"
);
}
#[test]
fn every_pinned_mutation_breaks_verification() {
let v = vectors();
let private_key = hex_of(&v, &["signer", "ed25519_private_key_hex"]);
let signer = Signer::from_key_bytes(&private_key).expect("valid key material");
let public_key = signer.public_key_bytes();
let good_signature = v["vector"]["expected_signature_hex"].as_str().unwrap();
let base = {
let mut base = vector_request(&v);
base.asserted_approval
.as_option_mut()
.unwrap()
.signature_hex = good_signature.to_owned();
base
};
let cases = v["must_not_verify"].as_array().expect("a list of cases");
assert_eq!(cases.len(), 6, "every pinned case must be exercised");
for case in cases {
let id = case["id"].as_str().unwrap();
let mut request = base.clone();
match id {
"tampered-responder" => {
request
.asserted_approval
.as_option_mut()
.unwrap()
.responder
.as_option_mut()
.unwrap()
.external_id = "U0EVE".to_owned();
}
"tampered-decision" => {
request.decision = Some(Decision::Deny(Box::new(Deny {
reason: "approved by Ada".to_owned(),
..Default::default()
})));
}
"tampered-modified-args" => {
let Some(Decision::Approve(approve)) = request.decision.as_mut() else {
panic!("the vector's decision is an Approve");
};
approve.modified_args_json = r#"{"path":"/etc/shadow"}"#.to_owned();
}
"tampered-resolve-token" => {
request.resolve_token = "resolve-token-conformance-2".to_owned();
}
"cross-domain-edge-identity" | "malformed-signature-hex" => {
}
other => panic!("unhandled conformance case {other}"),
}
assert_eq!(
crate::hex::lower(&canonical_bytes(&request)),
case["canonical_bytes_hex"].as_str().unwrap(),
"case {id}: the mutation applied here must be the one the vector pinned"
);
let signatures: Vec<String> = case["signature_hex"].as_str().map_or_else(
|| {
case["signature_hex_values"]
.as_array()
.expect("one of signature_hex / signature_hex_values")
.iter()
.map(|s| s.as_str().unwrap().to_owned())
.collect()
},
|s| vec![s.to_owned()],
);
for signature in signatures {
request
.asserted_approval
.as_option_mut()
.unwrap()
.signature_hex = signature.clone();
assert!(
!verify_approval_assertion([public_key.as_slice()], &request),
"case {id}: signature {signature:?} must not verify"
);
}
}
}
}