use serde::{Deserialize, Serialize};
use std::fmt;
use std::path::{Path, PathBuf};
use std::sync::{Arc, OnceLock};
use tokio::sync::{OwnedSemaphorePermit, Semaphore};
pub type Result<T> = std::result::Result<T, AsxError>;
pub fn escape_xml(s: &str) -> String {
let mut out = String::new();
let mut last = 0usize;
let mut modified = false;
let mut stripped_count: usize = 0;
for (i, b) in s.bytes().enumerate() {
let escaped = match b {
0x00..=0x08 | 0x0B..=0x0C | 0x0E..=0x1F | 0x7F => {
if !modified {
out.reserve(s.len());
modified = true;
}
out.push_str(&s[last..i]);
last = i + 1;
stripped_count += 1;
continue;
}
b'&' => "&",
b'<' => "<",
b'>' => ">",
b'"' => """,
b'\'' => "'",
_ => continue,
};
if !modified {
out.reserve(s.len() + 16);
modified = true;
}
out.push_str(&s[last..i]);
out.push_str(escaped);
last = i + 1;
}
if !modified {
return s.to_owned();
}
out.push_str(&s[last..]);
if stripped_count > 0 {
tracing::warn!(
stripped_bytes = stripped_count,
"escape_xml stripped {} forbidden XML 1.0 control character(s) from input; \
output differs from input — check the source field for binary/control data",
stripped_count,
);
}
out
}
fn default_blocking_crypto_concurrency() -> usize {
std::thread::available_parallelism()
.map(|n| n.get().saturating_mul(2))
.unwrap_or(8)
.clamp(4, 128)
}
const BLOCKING_CRYPTO_CONCURRENCY_ENV: &str = "ASX_BLOCKING_CRYPTO_CONCURRENCY";
fn configured_blocking_crypto_concurrency() -> usize {
std::env::var(BLOCKING_CRYPTO_CONCURRENCY_ENV)
.ok()
.and_then(|raw| raw.trim().parse::<usize>().ok())
.filter(|value| *value > 0)
.map(|value| value.clamp(1, 4096))
.unwrap_or_else(default_blocking_crypto_concurrency)
}
fn blocking_crypto_semaphore() -> Arc<Semaphore> {
static SEM: OnceLock<Arc<Semaphore>> = OnceLock::new();
Arc::clone(
SEM.get_or_init(|| Arc::new(Semaphore::new(configured_blocking_crypto_concurrency()))),
)
}
pub const DEFAULT_MAX_BODY_BYTES: usize = 256 * 1024 * 1024;
#[derive(Clone, Debug)]
pub struct CryptoAdmissionControl {
semaphore: Arc<Semaphore>,
label: &'static str,
}
impl CryptoAdmissionControl {
pub fn new(concurrency: usize) -> Self {
let cap = concurrency.clamp(1, 4096);
Self {
semaphore: Arc::new(Semaphore::new(cap)),
label: "instance-scoped crypto semaphore",
}
}
pub fn process_global() -> Self {
Self {
semaphore: blocking_crypto_semaphore(),
label: "process-global crypto semaphore",
}
}
pub async fn acquire(
&self,
stage: &'static str,
session: &SessionContext,
) -> Result<OwnedSemaphorePermit> {
Arc::clone(&self.semaphore)
.acquire_owned()
.await
.map_err(|_| {
AsxError::new(
ErrorCode::TransportFailure,
format!("{} is closed", self.label),
ErrorContext::for_session(stage, session),
)
})
}
}
#[cfg(feature = "as4")]
pub(crate) fn bytes_to_utf8_str<'a>(
bytes: &'a [u8],
stage: &'static str,
session: &SessionContext,
) -> Result<&'a str> {
std::str::from_utf8(bytes).map_err(|_| {
AsxError::new(
ErrorCode::ParseFailed,
format!("{stage}: payload is not valid UTF-8"),
ErrorContext::for_session(stage, session),
)
})
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ErrorContext {
pub stage: &'static str,
pub message_id: Option<String>,
pub partner_id: Option<String>,
pub session_id: Option<String>,
}
impl ErrorContext {
#[must_use]
pub fn new(stage: &'static str) -> Self {
Self {
stage,
message_id: None,
partner_id: None,
session_id: None,
}
}
#[must_use]
pub fn for_session(stage: &'static str, session: &SessionContext) -> Self {
Self::new(stage).with_session_and_partner(session.session_id(), session.partner_id())
}
#[must_use]
pub fn for_session_with_message(
stage: &'static str,
session: &SessionContext,
message_id: impl Into<String>,
) -> Self {
Self::new(stage)
.with_session_and_partner(session.session_id(), session.partner_id())
.with_message_id(message_id)
}
#[must_use]
pub fn with_session_and_partner(
mut self,
session_id: impl Into<String>,
partner_id: impl Into<String>,
) -> Self {
self.session_id = Some(session_id.into());
self.partner_id = Some(partner_id.into());
self
}
#[must_use]
pub fn with_message_id(mut self, message_id: impl Into<String>) -> Self {
self.message_id = Some(message_id.into());
self
}
#[must_use]
pub fn with_partner_id(mut self, partner_id: impl Into<String>) -> Self {
self.partner_id = Some(partner_id.into());
self
}
#[must_use]
pub fn with_session_id(mut self, session_id: impl Into<String>) -> Self {
self.session_id = Some(session_id.into());
self
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum ErrorCode {
InvalidInput,
ParseFailed,
SecurityVerificationFailed,
DecryptionFailed,
PolicyViolation,
TransportFailure,
InteropViolation,
ReliabilityFailure,
NotFound,
CapacityExhausted,
PayloadTooLarge,
StorageBackendFailure,
CertificateRevoked,
CertificateExpired,
Timeout,
}
impl ErrorCode {
pub fn as_str(self) -> &'static str {
match self {
Self::InvalidInput => "invalid_input",
Self::ParseFailed => "parse_failed",
Self::SecurityVerificationFailed => "security_verification_failed",
Self::DecryptionFailed => "decryption_failed",
Self::PolicyViolation => "policy_violation",
Self::TransportFailure => "transport_failure",
Self::InteropViolation => "interop_violation",
Self::ReliabilityFailure => "reliability_failure",
Self::NotFound => "not_found",
Self::CapacityExhausted => "capacity_exhausted",
Self::PayloadTooLarge => "payload_too_large",
Self::StorageBackendFailure => "storage_backend_failure",
Self::CertificateRevoked => "certificate_revoked",
Self::CertificateExpired => "certificate_expired",
Self::Timeout => "timeout",
}
}
pub fn to_http_status(self) -> u16 {
match self {
Self::InvalidInput => 400,
Self::ParseFailed => 400,
Self::SecurityVerificationFailed => 401,
Self::DecryptionFailed => 400,
Self::PolicyViolation => 422,
Self::TransportFailure => 502,
Self::InteropViolation => 400,
Self::ReliabilityFailure => 503,
Self::NotFound => 404,
Self::CapacityExhausted => 429,
Self::PayloadTooLarge => 413,
Self::StorageBackendFailure => 503,
Self::CertificateRevoked => 403,
Self::CertificateExpired => 403,
Self::Timeout => 504,
}
}
pub fn remediation_hint(self) -> Option<&'static str> {
match self {
Self::DecryptionFailed => Some(
"Verify that the recipient certificate PEM and its private key PEM match. \
Ensure the sender is encrypting to the correct public certificate. \
Re-key the key pair if the certificate has been re-issued.",
),
Self::SecurityVerificationFailed => Some(
"Confirm the trust anchor PEM includes the full CA chain of the signer. \
Check certificate validity period. \
Ensure CRL distribution points or OCSP responders are reachable.",
),
Self::TransportFailure => Some(
"Check network connectivity and DNS resolution for the remote endpoint. \
Verify TLS certificate chain and mutual-TLS configuration. \
Ensure the spool directory exists and is writable.",
),
Self::ReliabilityFailure => Some(
"Ensure an EventBus broadcast subscriber is active before message sends. \
Check dedup and reconciliation backend availability and capacity.",
),
Self::PolicyViolation => Some(
"Review the PMode and profile configuration against the partner specification. \
Verify the interop mode matches the partner's published requirements.",
),
Self::CapacityExhausted => Some(
"Shed load or retry after a backoff delay. \
Consider increasing channel capacity or conversation gate limits.",
),
Self::StorageBackendFailure => Some(
"Check dedup/reconciliation/audit backend connectivity and disk space. \
Inspect backend logs for I/O errors. \
Consider a circuit-breaker or fallback backend for resilience.",
),
Self::CertificateRevoked => Some(
"The partner's signing certificate has been revoked by its issuing CA. \
Contact the trading partner to obtain a replacement certificate. \
Update the trust anchor and retry.",
),
Self::CertificateExpired => Some(
"The partner's signing certificate has passed its notAfter validity date. \
Request a renewed certificate from the trading partner. \
Do not extend trust to expired certificates.",
),
Self::Timeout => Some(
"The remote endpoint did not respond within the configured timeout. \
Verify network connectivity and DNS resolution. \
Apply exponential back-off before retrying.",
),
_ => None,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AsxError {
pub code: ErrorCode,
pub message: String,
pub context: Box<ErrorContext>,
}
impl AsxError {
pub fn new(code: ErrorCode, message: impl Into<String>, context: ErrorContext) -> Self {
Self {
code,
message: message.into(),
context: Box::new(context),
}
}
pub fn remediation_hint(&self) -> Option<&'static str> {
self.code.remediation_hint()
}
#[must_use]
pub fn with_partner_id(mut self, partner_id: impl Into<String>) -> Self {
self.context = Box::new(self.context.with_partner_id(partner_id));
self
}
#[must_use]
pub fn with_session_id(mut self, session_id: impl Into<String>) -> Self {
self.context = Box::new(self.context.with_session_id(session_id));
self
}
#[must_use]
pub fn with_session_and_partner(
mut self,
session_id: impl Into<String>,
partner_id: impl Into<String>,
) -> Self {
self.context = Box::new(
self.context
.with_session_and_partner(session_id, partner_id),
);
self
}
#[must_use]
pub fn with_message_id(mut self, message_id: impl Into<String>) -> Self {
self.context = Box::new(self.context.with_message_id(message_id));
self
}
#[inline]
pub fn is_duplicate(&self) -> bool {
self.code == ErrorCode::ReliabilityFailure && self.message.contains("replay")
}
#[inline]
pub fn is_storage_failure(&self) -> bool {
self.code == ErrorCode::StorageBackendFailure
}
}
impl fmt::Display for AsxError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(
f,
"{} [{}|stage={}]",
self.message,
self.code.as_str(),
self.context.stage
)?;
if let Some(pid) = &self.context.partner_id {
write!(f, "[partner={pid}]")?;
}
if let Some(mid) = &self.context.message_id {
write!(f, "[msg={mid}]")?;
}
if let Some(sid) = &self.context.session_id {
write!(f, "[session={sid}]")?;
}
Ok(())
}
}
impl std::error::Error for AsxError {}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
#[non_exhaustive]
pub enum InteropMode {
#[default]
Strict,
#[cfg(feature = "interop-relaxed")]
Relaxed,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SessionContext {
session_id: String,
partner_id: String,
profile_name: String,
metadata: SessionMetadata,
cert_handle: Arc<CertHandle>,
correlation_scope: CorrelationScope,
#[cfg(any(feature = "as2", feature = "as4"))]
trust_anchors_cache: TrustAnchorCache,
#[cfg(any(feature = "as2", feature = "as4"))]
x509_store_cache: X509StoreCache,
}
#[cfg(feature = "as4")]
pub(crate) fn decode_xml_base64(value: &str, label: &str, stage: &'static str) -> Result<Vec<u8>> {
use base64::Engine as _;
let normalized: String = value.chars().filter(|c| !c.is_ascii_whitespace()).collect();
base64::engine::general_purpose::STANDARD
.decode(normalized)
.map_err(|err| {
AsxError::new(
ErrorCode::ParseFailed,
format!("failed to decode base64 {label}: {err}"),
ErrorContext::new(stage),
)
})
}
#[must_use]
pub(crate) fn redact_present(present: bool) -> &'static str {
if present { "<redacted>" } else { "<none>" }
}
#[must_use]
#[inline]
pub fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
use subtle::ConstantTimeEq;
if a.len() != b.len() {
return false;
}
a.ct_eq(b).into()
}
#[derive(Debug, Clone, PartialEq, Eq, Default)]
#[non_exhaustive]
pub struct SessionMetadata {
pub effective_policy_snapshot_json: Option<String>,
pub(crate) strict_runtime_bootstrap_validated: bool,
}
#[derive(Debug, Clone)]
pub struct SessionContextBuilder {
session_id: String,
partner_id: String,
profile_name: String,
cert_handle: Option<CertHandle>,
effective_policy_snapshot_json: Option<String>,
correlation_scope: Option<CorrelationScope>,
}
impl SessionContextBuilder {
pub fn new(session_id: impl Into<String>, partner_id: impl Into<String>) -> Self {
Self {
session_id: session_id.into(),
partner_id: partner_id.into(),
profile_name: "strict".to_string(),
cert_handle: None,
effective_policy_snapshot_json: None,
correlation_scope: None,
}
}
pub fn profile_name(mut self, profile_name: impl Into<String>) -> Self {
self.profile_name = profile_name.into();
self
}
pub fn cert_handle(mut self, cert_handle: CertHandle) -> Self {
self.cert_handle = Some(cert_handle);
self
}
fn default_cert_handle_key_id(&self) -> String {
format!("cert:{}", self.partner_id)
}
fn cert_handle_or_init(&mut self) -> &mut CertHandle {
let key_id = self.default_cert_handle_key_id();
self.cert_handle
.get_or_insert_with(|| CertHandle::new(key_id))
}
pub fn with_trust_anchor_pem(mut self, pem: impl Into<String>) -> Self {
self.cert_handle_or_init()
.trust_anchor_pems
.push(pem.into());
self
}
pub fn with_intermediate_ca_pem(mut self, pem: impl Into<String>) -> Self {
self.cert_handle_or_init()
.intermediate_ca_pems
.push(pem.into());
self
}
pub fn with_intermediate_ca_pems(
mut self,
pems: impl IntoIterator<Item = impl Into<String>>,
) -> Self {
self.cert_handle_or_init()
.intermediate_ca_pems
.extend(pems.into_iter().map(|p| p.into()));
self
}
pub fn with_trust_anchor_pems(
mut self,
pems: impl IntoIterator<Item = impl Into<String>>,
) -> Self {
self.cert_handle_or_init()
.trust_anchor_pems
.extend(pems.into_iter().map(|p| p.into()));
self
}
pub fn with_ocsp_mode(mut self, mode: OcspMode) -> Self {
self.cert_handle_or_init().ocsp_mode = mode;
self
}
pub fn with_signing_cert_pem(mut self, pem: impl Into<String>) -> Self {
self.cert_handle_or_init().signing_cert_pem = Some(pem.into());
self
}
pub fn with_signing_key_pem(mut self, pem: impl Into<String>) -> Self {
self.cert_handle_or_init().signing_key_pem = Some(zeroize::Zeroizing::new(pem.into()));
self
}
pub fn with_recipient_cert_pem(mut self, pem: impl Into<String>) -> Self {
self.cert_handle_or_init().recipient_cert_pem = Some(pem.into());
self
}
pub fn with_signing_material(
mut self,
cert_pem: impl Into<String>,
key_pem: impl Into<String>,
) -> Self {
let ch = self.cert_handle_or_init();
ch.signing_cert_pem = Some(cert_pem.into());
ch.signing_key_pem = Some(zeroize::Zeroizing::new(key_pem.into()));
self
}
pub fn with_fingerprint_sha256(mut self, fingerprint: impl Into<String>) -> Self {
self.cert_handle_or_init().fingerprint_sha256 = fingerprint.into();
self
}
pub fn effective_policy_snapshot_json(mut self, snapshot_json: impl Into<String>) -> Self {
self.effective_policy_snapshot_json = Some(snapshot_json.into());
self
}
pub fn correlation_scope(
mut self,
root_id: impl Into<String>,
parent_message_id: Option<String>,
) -> Self {
self.correlation_scope = Some(CorrelationScope {
root_id: root_id.into(),
parent_message_id,
traceparent: None,
});
self
}
pub fn build(self) -> Result<SessionContext> {
let mut session = SessionContext::new(self.session_id, self.partner_id, self.profile_name)?;
if let Some(cert_handle) = self.cert_handle {
match (&cert_handle.signing_key_pem, &cert_handle.signing_cert_pem) {
(Some(_), None) | (None, Some(_)) => {
return Err(AsxError::new(
ErrorCode::InvalidInput,
"signing_key_pem and signing_cert_pem must both be set or both absent",
ErrorContext::new("session_context_builder"),
));
}
_ => {}
}
#[cfg(any(feature = "as2", feature = "as4"))]
validate_cert_handle_outbound_pem(&cert_handle)?;
session = session.with_cert_handle(cert_handle)?;
}
if let Some(correlation_scope) = self.correlation_scope {
if correlation_scope.root_id.trim().is_empty() {
return Err(AsxError::new(
ErrorCode::InvalidInput,
"correlation root_id must not be empty",
ErrorContext::for_session("session_context_builder", &session),
));
}
session.correlation_scope = correlation_scope;
}
if let Some(snapshot_json) = self.effective_policy_snapshot_json {
session = session.with_effective_policy_snapshot_json(snapshot_json)?;
}
Ok(session)
}
}
#[cfg(any(feature = "as2", feature = "as4"))]
fn validate_cert_handle_outbound_pem(cert_handle: &CertHandle) -> Result<()> {
if let (Some(key_pem), Some(cert_pem)) =
(&cert_handle.signing_key_pem, &cert_handle.signing_cert_pem)
{
let cert = openssl::x509::X509::from_pem(cert_pem.as_bytes()).map_err(|_| {
AsxError::new(
ErrorCode::InvalidInput,
"signing_cert_pem is not a valid PEM X.509 certificate",
ErrorContext::new("session_context_builder_validate"),
)
})?;
let key = openssl::pkey::PKey::private_key_from_pem(key_pem.as_bytes()).map_err(|_| {
AsxError::new(
ErrorCode::InvalidInput,
"signing_key_pem is not a valid PEM private key",
ErrorContext::new("session_context_builder_validate"),
)
})?;
let cert_pub = cert.public_key().map_err(|_| {
AsxError::new(
ErrorCode::InvalidInput,
"signing_cert_pem does not contain a usable public key",
ErrorContext::new("session_context_builder_validate"),
)
})?;
if !key.public_eq(&cert_pub) {
return Err(AsxError::new(
ErrorCode::InvalidInput,
"signing_key_pem does not match signing_cert_pem",
ErrorContext::new("session_context_builder_validate"),
));
}
}
if let Some(pem) = &cert_handle.recipient_cert_pem {
openssl::x509::X509::from_pem(pem.as_bytes()).map_err(|_| {
AsxError::new(
ErrorCode::InvalidInput,
"recipient_cert_pem is not a valid PEM X.509 certificate",
ErrorContext::new("session_context_builder_validate"),
)
})?;
}
Ok(())
}
#[derive(Debug, Default, Clone)]
#[cfg(any(feature = "as2", feature = "as4"))]
pub(crate) struct TrustAnchorCache(Arc<OnceLock<Vec<openssl::x509::X509>>>);
#[cfg(any(feature = "as2", feature = "as4"))]
impl PartialEq for TrustAnchorCache {
fn eq(&self, _: &Self) -> bool {
true }
}
#[cfg(any(feature = "as2", feature = "as4"))]
impl Eq for TrustAnchorCache {}
#[derive(Default, Clone)]
#[cfg(any(feature = "as2", feature = "as4"))]
pub(crate) struct X509StoreCache(Arc<OnceLock<Arc<openssl::x509::store::X509Store>>>);
#[cfg(any(feature = "as2", feature = "as4"))]
impl std::fmt::Debug for X509StoreCache {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_tuple("X509StoreCache")
.field(&self.0.get().is_some())
.finish()
}
}
#[cfg(any(feature = "as2", feature = "as4"))]
impl PartialEq for X509StoreCache {
fn eq(&self, _: &Self) -> bool {
true
}
}
#[cfg(any(feature = "as2", feature = "as4"))]
impl Eq for X509StoreCache {}
#[derive(Clone, PartialEq, Eq)]
pub struct CertHandle {
pub key_id: String,
pub fingerprint_sha256: String,
pub trust_anchor_pems: Vec<String>,
pub intermediate_ca_pems: Vec<String>,
pub revocation_crl_pems: Vec<String>,
pub ocsp_mode: OcspMode,
pub ocsp_failure_mode: OcspFailureMode,
pub stapled_ocsp_responses_der: Vec<Vec<u8>>,
pub responder_ocsp_responses_der: Vec<Vec<u8>>,
pub signing_cert_pem: Option<String>,
pub signing_key_pem: Option<zeroize::Zeroizing<String>>,
pub recipient_cert_pem: Option<String>,
}
impl std::fmt::Debug for CertHandle {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("CertHandle")
.field("key_id", &self.key_id)
.field("fingerprint_sha256", &self.fingerprint_sha256)
.field("trust_anchor_pems", &self.trust_anchor_pems)
.field("intermediate_ca_pems", &self.intermediate_ca_pems)
.field("revocation_crl_pems", &self.revocation_crl_pems)
.field("ocsp_mode", &self.ocsp_mode)
.field("ocsp_failure_mode", &self.ocsp_failure_mode)
.field(
"stapled_ocsp_responses_der",
&self.stapled_ocsp_responses_der,
)
.field(
"responder_ocsp_responses_der",
&self.responder_ocsp_responses_der,
)
.field("signing_cert_pem", &self.signing_cert_pem)
.field(
"signing_key_pem",
&self.signing_key_pem.as_ref().map(|_| "<redacted>"),
)
.field("recipient_cert_pem", &self.recipient_cert_pem)
.finish()
}
}
impl CertHandle {
pub fn new(key_id: impl Into<String>) -> Self {
Self {
key_id: key_id.into(),
fingerprint_sha256: String::new(),
trust_anchor_pems: Vec::new(),
intermediate_ca_pems: Vec::new(),
revocation_crl_pems: Vec::new(),
ocsp_mode: OcspMode::default(),
ocsp_failure_mode: OcspFailureMode::HardFail,
stapled_ocsp_responses_der: Vec::new(),
responder_ocsp_responses_der: Vec::new(),
signing_cert_pem: None,
signing_key_pem: None,
recipient_cert_pem: None,
}
}
pub fn set_signing_key_pem(&mut self, key_pem: impl Into<String>) {
self.signing_key_pem = Some(zeroize::Zeroizing::new(key_pem.into()));
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
#[non_exhaustive]
pub enum OcspMode {
Disabled,
StapledOnly,
#[default]
ResponderOnly,
StapledThenResponder,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
#[non_exhaustive]
pub enum OcspFailureMode {
#[default]
HardFail,
SoftFail,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CorrelationScope {
pub root_id: String,
pub parent_message_id: Option<String>,
pub traceparent: Option<Arc<str>>,
}
#[non_exhaustive]
#[derive(Debug)]
pub enum PayloadInput<'a> {
Owned(Vec<u8>),
Shared(Arc<[u8]>),
Borrowed(&'a [u8]),
}
impl<'a> PayloadInput<'a> {
pub fn as_slice(&self) -> &[u8] {
match self {
Self::Owned(payload) => payload,
Self::Shared(payload) => payload,
Self::Borrowed(payload) => payload,
}
}
pub fn into_arc(self) -> Arc<[u8]> {
match self {
Self::Owned(payload) => Arc::from(payload),
Self::Shared(payload) => payload,
Self::Borrowed(payload) => Arc::from(payload),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum SpoolEncryption {
Plaintext,
Aes256Gcm { key: Arc<[u8; 32]> },
}
pub(crate) const SPOOLED_AES256_GCM_MAGIC: [u8; 8] = *b"ASXSPG01";
pub(crate) const SPOOLED_AES256_GCM_NONCE_LEN: usize = 12;
pub(crate) const SPOOLED_AES256_GCM_TAG_LEN: usize = 16;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SpoolLifecyclePolicy {
pub delete_on_materialize: bool,
pub secure_delete_on_materialize: bool,
}
impl Default for SpoolLifecyclePolicy {
fn default() -> Self {
Self {
delete_on_materialize: true,
secure_delete_on_materialize: false,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum ReceivedBodyHandle {
InMemory(Arc<[u8]>),
Spooled {
path: PathBuf,
encryption: SpoolEncryption,
lifecycle: SpoolLifecyclePolicy,
},
}
impl ReceivedBodyHandle {
#[must_use]
pub fn from_payload_input(input: PayloadInput<'_>) -> Self {
Self::InMemory(input.into_arc())
}
pub fn payload_len(&self, stage: &'static str, session: &SessionContext) -> Result<usize> {
match self {
Self::InMemory(bytes) => Ok(bytes.len()),
Self::Spooled { path, .. } => {
let metadata = std::fs::metadata(path).map_err(|err| {
AsxError::new(
ErrorCode::TransportFailure,
format!("failed to stat spooled body {}: {err}", path.display()),
ErrorContext::for_session(stage, session),
)
})?;
usize::try_from(metadata.len()).map_err(|_| {
AsxError::new(
ErrorCode::PolicyViolation,
format!(
"spooled body {} exceeds platform addressable size",
path.display()
),
ErrorContext::for_session(stage, session),
)
})
}
}
}
pub fn materialize_contiguous(
&self,
stage: &'static str,
session: &SessionContext,
) -> Result<Arc<[u8]>> {
match self {
Self::InMemory(bytes) => Ok(Arc::clone(bytes)),
Self::Spooled {
path, encryption, ..
} => Ok(Arc::from(read_spooled_bytes(
path, encryption, stage, session,
)?)),
}
}
pub fn into_arc(self, stage: &'static str, session: &SessionContext) -> Result<Arc<[u8]>> {
match self {
Self::InMemory(bytes) => Ok(bytes),
Self::Spooled {
path,
encryption,
lifecycle,
} => {
let bytes = read_spooled_bytes(&path, &encryption, stage, session)?;
if lifecycle.delete_on_materialize {
delete_spooled_file(
&path,
lifecycle.secure_delete_on_materialize,
stage,
session,
)?;
}
Ok(Arc::from(bytes))
}
}
}
pub fn dispose(self, stage: &'static str, session: &SessionContext) -> Result<()> {
match self {
Self::InMemory(_) => Ok(()),
Self::Spooled {
path, lifecycle, ..
} => {
if lifecycle.delete_on_materialize {
delete_spooled_file(
&path,
lifecycle.secure_delete_on_materialize,
stage,
session,
)?;
}
Ok(())
}
}
}
}
fn read_spooled_bytes(
path: &Path,
encryption: &SpoolEncryption,
stage: &'static str,
session: &SessionContext,
) -> Result<Vec<u8>> {
let bytes = std::fs::read(path).map_err(|err| {
AsxError::new(
ErrorCode::TransportFailure,
format!("failed to read spooled body {}: {err}", path.display()),
ErrorContext::for_session(stage, session),
)
})?;
match encryption {
SpoolEncryption::Plaintext => Ok(bytes),
SpoolEncryption::Aes256Gcm { key } => {
let min_len = SPOOLED_AES256_GCM_MAGIC.len()
+ SPOOLED_AES256_GCM_NONCE_LEN
+ SPOOLED_AES256_GCM_TAG_LEN;
if bytes.len() < min_len {
return Err(AsxError::new(
ErrorCode::DecryptionFailed,
format!(
"spooled encrypted body {} is too short for AES-GCM envelope",
path.display()
),
ErrorContext::for_session(stage, session),
));
}
let magic = &bytes[..SPOOLED_AES256_GCM_MAGIC.len()];
if magic != SPOOLED_AES256_GCM_MAGIC {
return Err(AsxError::new(
ErrorCode::DecryptionFailed,
format!(
"spooled encrypted body {} has invalid envelope magic",
path.display()
),
ErrorContext::for_session(stage, session),
));
}
let nonce_start = SPOOLED_AES256_GCM_MAGIC.len();
let nonce_end = nonce_start + SPOOLED_AES256_GCM_NONCE_LEN;
let tag_start = bytes.len() - SPOOLED_AES256_GCM_TAG_LEN;
let nonce = &bytes[nonce_start..nonce_end];
let ciphertext = &bytes[nonce_end..tag_start];
let tag = &bytes[tag_start..];
decrypt_spooled_aes256_gcm(path, key.as_ref(), nonce, ciphertext, tag, stage, session)
}
}
}
#[cfg(any(feature = "as2", feature = "as4", feature = "async-ocsp"))]
fn decrypt_spooled_aes256_gcm(
path: &Path,
key: &[u8],
nonce: &[u8],
ciphertext: &[u8],
tag: &[u8],
stage: &'static str,
session: &SessionContext,
) -> Result<Vec<u8>> {
openssl::symm::decrypt_aead(
openssl::symm::Cipher::aes_256_gcm(),
key,
Some(nonce),
&[],
ciphertext,
tag,
)
.map_err(|err| {
AsxError::new(
ErrorCode::DecryptionFailed,
format!(
"failed to decrypt spooled encrypted body {}: {err}",
path.display()
),
ErrorContext::for_session(stage, session),
)
})
}
#[cfg(not(any(feature = "as2", feature = "as4", feature = "async-ocsp")))]
fn decrypt_spooled_aes256_gcm(
path: &Path,
_key: &[u8],
_nonce: &[u8],
_ciphertext: &[u8],
_tag: &[u8],
stage: &'static str,
session: &SessionContext,
) -> Result<Vec<u8>> {
Err(AsxError::new(
ErrorCode::PolicyViolation,
format!(
"spool AES-256-GCM decryption unavailable for {} without crypto protocol features",
path.display()
),
ErrorContext::for_session(stage, session),
))
}
fn delete_spooled_file(
path: &Path,
secure_delete: bool,
stage: &'static str,
session: &SessionContext,
) -> Result<()> {
if secure_delete {
use std::io::{Seek, SeekFrom, Write};
let mut file = std::fs::OpenOptions::new()
.read(true)
.write(true)
.open(path)
.map_err(|err| {
AsxError::new(
ErrorCode::TransportFailure,
format!(
"failed to open spooled file {} for secure delete: {err}",
path.display()
),
ErrorContext::for_session(stage, session),
)
})?;
let file_len = file
.metadata()
.map_err(|err| {
AsxError::new(
ErrorCode::TransportFailure,
format!(
"failed to stat spooled file {} for secure delete: {err}",
path.display()
),
ErrorContext::for_session(stage, session),
)
})?
.len();
file.seek(SeekFrom::Start(0)).map_err(|err| {
AsxError::new(
ErrorCode::TransportFailure,
format!(
"failed to seek spooled file {} for secure delete: {err}",
path.display()
),
ErrorContext::for_session(stage, session),
)
})?;
let zeroes = vec![0u8; 8192];
let mut remaining = file_len;
while remaining > 0 {
let write_len =
usize::try_from(remaining.min(zeroes.len() as u64)).unwrap_or(zeroes.len());
file.write_all(&zeroes[..write_len]).map_err(|err| {
AsxError::new(
ErrorCode::TransportFailure,
format!(
"failed to overwrite spooled file {} for secure delete: {err}",
path.display()
),
ErrorContext::for_session(stage, session),
)
})?;
remaining -= write_len as u64;
}
file.flush().map_err(|err| {
AsxError::new(
ErrorCode::TransportFailure,
format!(
"failed to flush overwritten spooled file {}: {err}",
path.display()
),
ErrorContext::for_session(stage, session),
)
})?;
file.sync_all().map_err(|err| {
AsxError::new(
ErrorCode::TransportFailure,
format!(
"failed to sync overwritten spooled file {}: {err}",
path.display()
),
ErrorContext::for_session(stage, session),
)
})?;
}
std::fs::remove_file(path).map_err(|err| {
AsxError::new(
ErrorCode::TransportFailure,
format!("failed to remove spooled file {}: {err}", path.display()),
ErrorContext::for_session(stage, session),
)
})
}
impl SessionContext {
pub fn builder(
session_id: impl Into<String>,
partner_id: impl Into<String>,
) -> SessionContextBuilder {
SessionContextBuilder::new(session_id, partner_id)
}
pub fn new(
session_id: impl Into<String>,
partner_id: impl Into<String>,
profile_name: impl Into<String>,
) -> Result<Self> {
let session_id = session_id.into();
let partner_id = partner_id.into();
let profile_name = profile_name.into();
if session_id.trim().is_empty() {
return Err(AsxError::new(
ErrorCode::InvalidInput,
"session_id must not be empty",
ErrorContext::new("session_context_init"),
));
}
if partner_id.trim().is_empty() {
return Err(AsxError::new(
ErrorCode::InvalidInput,
"partner_id must not be empty",
ErrorContext::new("session_context_init").with_session_id(&session_id),
));
}
if profile_name.trim().is_empty() {
return Err(AsxError::new(
ErrorCode::InvalidInput,
"profile_name must not be empty",
ErrorContext::new("session_context_init")
.with_session_and_partner(&session_id, &partner_id),
));
}
Ok(Self {
metadata: SessionMetadata::default(),
cert_handle: Arc::new(CertHandle::new(format!("cert:{partner_id}"))),
correlation_scope: CorrelationScope {
root_id: format!("corr:{session_id}"),
parent_message_id: None,
traceparent: None,
},
session_id,
partner_id,
profile_name,
#[cfg(any(feature = "as2", feature = "as4"))]
trust_anchors_cache: TrustAnchorCache::default(),
#[cfg(any(feature = "as2", feature = "as4"))]
x509_store_cache: X509StoreCache::default(),
})
}
pub fn with_cert_handle(mut self, cert_handle: CertHandle) -> Result<Self> {
Self::validate_cert_handle_fields(&cert_handle, "session_context_cert_update", &self)?;
self.cert_handle = Arc::new(cert_handle);
#[cfg(any(feature = "as2", feature = "as4"))]
{
self.trust_anchors_cache = TrustAnchorCache::default();
self.x509_store_cache = X509StoreCache::default();
}
Ok(self)
}
pub fn rotate_cert_handle(&mut self, cert_handle: CertHandle) -> Result<()> {
Self::validate_cert_handle_fields(&cert_handle, "session_context_cert_rotate", self)?;
self.cert_handle = Arc::new(cert_handle);
#[cfg(any(feature = "as2", feature = "as4"))]
{
self.trust_anchors_cache = TrustAnchorCache::default();
self.x509_store_cache = X509StoreCache::default();
}
Ok(())
}
fn validate_cert_handle_fields(
cert_handle: &CertHandle,
stage: &'static str,
session: &SessionContext,
) -> Result<()> {
if cert_handle.key_id.trim().is_empty() {
return Err(AsxError::new(
ErrorCode::InvalidInput,
"cert handle key_id must not be empty",
ErrorContext::for_session(stage, session),
));
}
if cert_handle
.trust_anchor_pems
.iter()
.any(|pem| pem.trim().is_empty())
|| cert_handle
.revocation_crl_pems
.iter()
.any(|pem| pem.trim().is_empty())
|| cert_handle
.stapled_ocsp_responses_der
.iter()
.any(Vec::is_empty)
|| cert_handle
.responder_ocsp_responses_der
.iter()
.any(Vec::is_empty)
{
return Err(AsxError::new(
ErrorCode::InvalidInput,
"cert handle PKIX/OCSP material must not contain empty entries",
ErrorContext::for_session(stage, session),
));
}
Ok(())
}
pub fn with_effective_policy_snapshot_json(
mut self,
snapshot_json: impl Into<String>,
) -> Result<Self> {
let snapshot_json = snapshot_json.into();
if snapshot_json.trim().is_empty() {
return Err(AsxError::new(
ErrorCode::InvalidInput,
"effective policy snapshot JSON must not be empty",
ErrorContext::for_session("session_context_metadata_update", &self),
));
}
self.metadata.effective_policy_snapshot_json = Some(snapshot_json);
Ok(self)
}
pub fn effective_policy_snapshot_json(&self) -> Option<&str> {
self.metadata.effective_policy_snapshot_json.as_deref()
}
pub fn strict_runtime_bootstrap_validated(&self) -> bool {
self.metadata.strict_runtime_bootstrap_validated
}
#[cfg(feature = "testing")]
#[must_use]
pub fn test_only_mark_strict_runtime_bootstrap_validated(self) -> Self {
self.with_strict_runtime_bootstrap_validated(true)
}
pub(crate) fn with_strict_runtime_bootstrap_validated(mut self, validated: bool) -> Self {
self.metadata.strict_runtime_bootstrap_validated = validated;
self
}
pub fn session_id(&self) -> &str {
&self.session_id
}
pub fn partner_id(&self) -> &str {
&self.partner_id
}
pub fn profile_name(&self) -> &str {
&self.profile_name
}
pub fn cert_handle(&self) -> &CertHandle {
self.cert_handle.as_ref()
}
#[cfg(any(feature = "as2", feature = "as4"))]
pub(crate) fn trust_anchors_x509(&self) -> Result<Vec<openssl::x509::X509>> {
if let Some(anchors) = self.trust_anchors_cache.0.get() {
return Ok(anchors.clone());
}
let mut anchors = Vec::new();
for pem in &self.cert_handle.trust_anchor_pems {
let certs = openssl::x509::X509::stack_from_pem(pem.as_bytes()).map_err(|e| {
AsxError::new(
ErrorCode::InvalidInput,
format!("invalid trust-anchor PEM in CertHandle: {e}"),
ErrorContext::new("session_parse_trust_anchors"),
)
})?;
anchors.extend(certs);
}
let _ = self.trust_anchors_cache.0.set(anchors.clone());
Ok(anchors)
}
#[cfg(any(feature = "as2", feature = "as4"))]
pub(crate) fn trust_anchor_x509_store(&self) -> Result<Arc<openssl::x509::store::X509Store>> {
if let Some(store) = self.x509_store_cache.0.get() {
return Ok(Arc::clone(store));
}
let anchors = self.trust_anchors_x509()?;
let mut builder = openssl::x509::store::X509StoreBuilder::new().map_err(|e| {
AsxError::new(
ErrorCode::InvalidInput,
format!("failed to build X.509 trust store: {e}"),
ErrorContext::new("session_build_x509_store"),
)
})?;
for cert in &anchors {
builder.add_cert(cert.clone()).map_err(|e| {
AsxError::new(
ErrorCode::InvalidInput,
format!("failed to add trust anchor to X.509 store: {e}"),
ErrorContext::new("session_build_x509_store"),
)
})?;
}
let store = Arc::new(builder.build());
let _ = self.x509_store_cache.0.set(Arc::clone(&store));
Ok(store)
}
pub fn correlation_scope(&self) -> &CorrelationScope {
&self.correlation_scope
}
pub fn with_incoming_traceparent(mut self, traceparent: Option<&str>) -> Self {
if let Some(tp) = traceparent {
self.correlation_scope.traceparent = Some(Arc::from(tp));
}
self
}
#[cfg(any(test, feature = "testing"))]
pub fn for_testing(session_id: impl Into<String>, partner_id: impl Into<String>) -> Self {
let session_id = session_id.into();
let partner_id = partner_id.into();
Self {
metadata: SessionMetadata::default(),
cert_handle: Arc::new(CertHandle {
ocsp_mode: OcspMode::Disabled,
ocsp_failure_mode: OcspFailureMode::SoftFail,
..CertHandle::new(format!("cert:{partner_id}"))
}),
correlation_scope: CorrelationScope {
root_id: format!("corr:{session_id}"),
parent_message_id: None,
traceparent: None,
},
session_id,
partner_id,
profile_name: "test".into(),
#[cfg(any(feature = "as2", feature = "as4"))]
trust_anchors_cache: TrustAnchorCache::default(),
#[cfg(any(feature = "as2", feature = "as4"))]
x509_store_cache: X509StoreCache::default(),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn escape_xml_prevents_injection() {
assert_eq!(escape_xml("A&B"), "A&B");
assert_eq!(escape_xml("A<B"), "A<B");
assert_eq!(escape_xml("A>B"), "A>B");
assert_eq!(escape_xml("A\"B"), "A"B");
assert_eq!(
escape_xml("msg<inject>B&C\"D"),
"msg<inject>B&C"D"
);
assert_eq!(escape_xml(""), "");
assert_eq!(escape_xml("hello-world"), "hello-world");
assert_eq!(escape_xml("ab\x00cd"), "abcd");
assert_eq!(escape_xml("\x01\x08\x0B\x0C\x0E\x1F\x7F"), "");
assert_eq!(escape_xml("a\x00<b\x00>"), "a<b>");
}
#[test]
fn error_code_strings_are_stable() {
assert_eq!(ErrorCode::TransportFailure.as_str(), "transport_failure");
assert_eq!(ErrorCode::InteropViolation.as_str(), "interop_violation");
}
#[test]
fn session_context_validation_rejects_empty_values() {
assert!(SessionContext::new("", "p", "strict").is_err());
assert!(SessionContext::new("s", "", "strict").is_err());
assert!(SessionContext::new("s", "p", "").is_err());
}
#[test]
fn session_context_has_deterministic_default_handles() {
let session = SessionContext::new("s1", "partner-a", "strict").expect("session");
assert_eq!(session.cert_handle().key_id, "cert:partner-a");
assert_eq!(session.correlation_scope().root_id, "corr:s1");
assert!(session.effective_policy_snapshot_json().is_none());
}
#[test]
fn session_context_builder_supports_incremental_configuration() {
let cert = CertHandle {
trust_anchor_pems: vec!["anchor-pem".into()],
..CertHandle::new("partner-key")
};
let session = SessionContext::builder("s-builder", "partner-z")
.profile_name("peppol")
.cert_handle(cert)
.effective_policy_snapshot_json("{\"mode\":\"Strict\"}")
.correlation_scope("corr-custom", Some("parent-1".into()))
.build()
.expect("builder session");
assert_eq!(session.session_id(), "s-builder");
assert_eq!(session.partner_id(), "partner-z");
assert_eq!(session.profile_name(), "peppol");
assert_eq!(session.cert_handle().key_id, "partner-key");
assert_eq!(session.correlation_scope().root_id, "corr-custom");
assert_eq!(
session.correlation_scope().parent_message_id.as_deref(),
Some("parent-1")
);
assert_eq!(
session.effective_policy_snapshot_json(),
Some("{\"mode\":\"Strict\"}")
);
}
#[test]
fn session_context_builder_rejects_blank_correlation_root() {
let err = SessionContext::builder("s-builder", "partner-z")
.correlation_scope(" ", None)
.build()
.expect_err("must reject blank correlation root");
assert_eq!(err.code, ErrorCode::InvalidInput);
}
#[test]
fn session_context_metadata_attaches_snapshot_json() {
let session = SessionContext::new("s1", "partner-a", "strict")
.expect("session")
.with_effective_policy_snapshot_json("{\"resolved_mode\":\"Strict\"}")
.expect("snapshot json");
assert_eq!(
session.effective_policy_snapshot_json(),
Some("{\"resolved_mode\":\"Strict\"}")
);
}
#[test]
fn session_context_metadata_rejects_empty_snapshot_json() {
let err = SessionContext::new("s1", "partner-a", "strict")
.expect("session")
.with_effective_policy_snapshot_json(" ")
.expect_err("must reject blank snapshot");
assert_eq!(err.code, ErrorCode::InvalidInput);
}
#[test]
fn cert_handle_new_has_expected_defaults() {
let h = CertHandle::new("my-key");
assert_eq!(h.key_id, "my-key");
assert!(h.trust_anchor_pems.is_empty());
assert_eq!(h.ocsp_mode, OcspMode::ResponderOnly);
assert_eq!(h.ocsp_failure_mode, OcspFailureMode::HardFail);
}
#[test]
fn cert_handle_struct_update_syntax_works() {
let base = CertHandle::new("base-key");
let updated = CertHandle {
trust_anchor_pems: vec!["fake-pem".into()],
ocsp_mode: OcspMode::Disabled,
..base
};
assert_eq!(updated.key_id, "base-key");
assert_eq!(updated.trust_anchor_pems, vec!["fake-pem".to_string()]);
assert_eq!(updated.ocsp_mode, OcspMode::Disabled);
}
#[test]
fn rotate_cert_handle_preserves_session_identity() {
let mut session =
SessionContext::new("rotate-session", "partner-b", "strict").expect("session");
let original_session_id = session.session_id().to_string();
let original_partner_id = session.partner_id().to_string();
let original_root_id = session.correlation_scope().root_id.clone();
let new_cert = CertHandle {
trust_anchor_pems: vec!["new-anchor-pem".into()],
..CertHandle::new("new-key")
};
session.rotate_cert_handle(new_cert).expect("rotate");
assert_eq!(session.session_id(), original_session_id);
assert_eq!(session.partner_id(), original_partner_id);
assert_eq!(session.correlation_scope().root_id, original_root_id);
assert_eq!(session.cert_handle().key_id, "new-key");
assert_eq!(
session.cert_handle().trust_anchor_pems,
vec!["new-anchor-pem".to_string()]
);
}
#[test]
fn rotate_cert_handle_rejects_empty_key_id() {
let mut session = SessionContext::new("s1", "p1", "strict").expect("session");
let bad_cert = CertHandle::new("");
assert!(session.rotate_cert_handle(bad_cert).is_err());
}
#[test]
fn arc_cert_handle_clone_shares_same_pointer() {
let session = SessionContext::new("s1", "p1", "strict").expect("session");
let clone = session.clone();
assert!(Arc::ptr_eq(&session.cert_handle, &clone.cert_handle));
}
#[test]
#[cfg(any(feature = "as2", feature = "as4"))]
fn with_cert_handle_resets_trust_anchor_cache() {
let session = SessionContext::new("s-cache", "partner-cache", "strict").expect("session");
assert!(session.trust_anchors_cache.0.get().is_none());
let new_handle = CertHandle::new("partner-cache-cert");
let session = session.with_cert_handle(new_handle).expect("set handle");
assert!(session.trust_anchors_cache.0.get().is_none());
let handle2 = CertHandle {
trust_anchor_pems: vec!["some-pem".into()],
..CertHandle::new("partner-cache-cert-2")
};
let _ = session.with_cert_handle(handle2);
}
#[test]
fn builder_with_trust_anchor_pem_does_not_leave_empty_key_id() {
let result = SessionContextBuilder::new("s1", "partner-xyz")
.with_trust_anchor_pem("fake-pem")
.build();
assert!(result.is_ok(), "build() must not fail: {:?}", result);
let session = result.unwrap();
assert_eq!(
session.cert_handle().key_id,
"cert:partner-xyz",
"key_id should be auto-derived from partner_id"
);
}
#[test]
fn builder_with_signing_cert_and_key_pem_do_not_leave_empty_key_id() {
let builder =
SessionContextBuilder::new("s1", "partner-abc").with_signing_cert_pem("not-real-pem");
assert_eq!(
builder.cert_handle.as_ref().expect("handle").key_id,
"cert:partner-abc",
);
}
#[test]
fn builder_with_fingerprint_sha256_sets_field() {
let builder =
SessionContextBuilder::new("s1", "partner-fp").with_fingerprint_sha256("aabbcc");
assert_eq!(
builder
.cert_handle
.as_ref()
.expect("handle")
.fingerprint_sha256,
"aabbcc",
);
}
#[test]
fn builder_with_signing_material_sets_both_fields() {
let builder = SessionContextBuilder::new("s1", "partner-mat")
.with_signing_material("cert-pem-value", "key-pem-value");
let ch = builder.cert_handle.as_ref().expect("handle");
assert_eq!(ch.signing_cert_pem.as_deref(), Some("cert-pem-value"));
assert!(ch.signing_key_pem.is_some());
assert_eq!(
ch.signing_key_pem.as_ref().map(|s| s.as_str()),
Some("key-pem-value")
);
}
#[test]
fn cert_handle_set_signing_key_pem_avoids_zeroize_dep() {
let mut ch = CertHandle::new("key");
ch.set_signing_key_pem("my-private-key");
assert!(ch.signing_key_pem.is_some());
assert_eq!(
ch.signing_key_pem.as_ref().map(|s| s.as_str()),
Some("my-private-key")
);
}
}