exo_node/
crosschecked_anchor_http.rs

1// Copyright 2026 Exochain Foundation
2// SPDX-License-Identifier: Apache-2.0
3
4//! Fail-closed HTTP boundary for commitment-only CrossChecked recording.
5
6use std::{
7    collections::BTreeMap,
8    fmt,
9    path::{Path, PathBuf},
10    sync::Arc,
11};
12
13use axum::{
14    Json, Router,
15    body::Bytes,
16    extract::{DefaultBodyLimit, Path as AxumPath, State},
17    http::{HeaderMap, StatusCode, header},
18    response::{IntoResponse, Response},
19    routing::{get, post},
20};
21use exo_api::crosschecked_anchor::{ANCHOR_PATH, MAX_REQUEST_BODY_BYTES};
22use exo_core::types::{Hash256, PublicKey, Signature, Timestamp};
23use rusqlite::{Connection, OptionalExtension, TransactionBehavior, params};
24use serde_json::json;
25use thiserror::Error;
26use zeroize::{Zeroize, Zeroizing};
27
28use crate::crosschecked_anchor_store::{
29    AnchorConflictKind, AnchorNodeIdentity, AnchorRecordDisposition, AnchorStore,
30    AnchorStoreConfig, AnchorStoreError, DurableAnchorSigner, SignOnceError, SubmissionContext,
31};
32
33pub const CROSSCHECKED_ANCHOR_BEARER_ENV: &str = "EXOCHAIN_CROSSCHECKED_ANCHOR_BEARER_TOKEN";
34pub const CROSSCHECKED_ANCHOR_EXPECTED_AUDIENCE_ENV: &str =
35    "EXOCHAIN_CROSSCHECKED_ANCHOR_EXPECTED_AUDIENCE";
36pub const CROSSCHECKED_ANCHOR_INTERMEDIATE_DID_ENV: &str =
37    "EXOCHAIN_CROSSCHECKED_ANCHOR_INTERMEDIATE_DID";
38pub const CROSSCHECKED_ANCHOR_INTERMEDIATE_KEY_ID_ENV: &str =
39    "EXOCHAIN_CROSSCHECKED_ANCHOR_INTERMEDIATE_KEY_ID";
40pub const CROSSCHECKED_ANCHOR_INTERMEDIATE_PUBLIC_KEY_ENV: &str =
41    "EXOCHAIN_CROSSCHECKED_ANCHOR_INTERMEDIATE_PUBLIC_KEY_HEX";
42pub const CROSSCHECKED_ANCHOR_GOVERNANCE_FROST_PUBLIC_KEY_ENV: &str =
43    "EXOCHAIN_CROSSCHECKED_ANCHOR_GOVERNANCE_FROST_PUBLIC_KEY_HEX";
44pub const CROSSCHECKED_ANCHOR_GOVERNANCE_FROST_KEY_EPOCH_ENV: &str =
45    "EXOCHAIN_CROSSCHECKED_ANCHOR_GOVERNANCE_FROST_KEY_EPOCH";
46pub const CROSSCHECKED_ANCHOR_NODE_KEY_ID_ENV: &str = "EXOCHAIN_CROSSCHECKED_ANCHOR_NODE_KEY_ID";
47
48const SOURCE_CODE: &str = "crosschecked";
49const SIGNER_SCHEMA_VERSION: &str = "crosschecked_anchor_signer_v1";
50const BEARER_VERIFIER_CONTEXT: &str = "exochain.crosschecked.anchor.transport_bearer_verifier.v1";
51
52type Clock = Arc<dyn Fn() -> Result<Timestamp, String> + Send + Sync>;
53type SignFunction = Arc<dyn Fn(&[u8]) -> Signature + Send + Sync>;
54
55/// Domain-separated, fixed-size verifier retained instead of the transport
56/// bearer. CrossChecked owns the random 256-bit bearer; EXOCHAIN hashes it at
57/// startup and never places its plaintext in route or middleware state.
58#[derive(Clone, PartialEq, Eq)]
59pub struct CrossCheckedBearerVerifier([u8; 32]);
60
61impl fmt::Debug for CrossCheckedBearerVerifier {
62    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
63        formatter.write_str("CrossCheckedBearerVerifier([REDACTED])")
64    }
65}
66
67impl CrossCheckedBearerVerifier {
68    /// Derive a verifier from an exact lowercase-hex 256-bit bearer.
69    #[must_use]
70    pub fn from_bearer(bearer: &str) -> Option<Self> {
71        if !is_valid_bearer(bearer) {
72            return None;
73        }
74        Some(Self(blake3::derive_key(
75            BEARER_VERIFIER_CONTEXT,
76            bearer.as_bytes(),
77        )))
78    }
79
80    /// Hash a presented bearer under the same domain and compare only the
81    /// fixed-size verifiers in constant time.
82    #[must_use]
83    pub fn verifies(&self, bearer: &str) -> bool {
84        let mut candidate = blake3::derive_key(BEARER_VERIFIER_CONTEXT, bearer.as_bytes());
85        let matches = constant_time_eq(&candidate, &self.0);
86        candidate.zeroize();
87        matches
88    }
89}
90
91#[derive(Clone)]
92pub struct CrossCheckedAnchorHttpState {
93    store: AnchorStore,
94    signer: Arc<dyn DurableAnchorSigner>,
95    bearer_verifier: CrossCheckedBearerVerifier,
96    clock: Clock,
97}
98
99impl fmt::Debug for CrossCheckedAnchorHttpState {
100    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
101        formatter
102            .debug_struct("CrossCheckedAnchorHttpState")
103            .field("store", &self.store)
104            .field("signer_identity", &self.signer.identity())
105            .field("bearer_verifier", &"[REDACTED]")
106            .finish_non_exhaustive()
107    }
108}
109
110impl CrossCheckedAnchorHttpState {
111    /// Construct route state from a verifier after the raw bearer is dropped.
112    pub fn new(
113        store: AnchorStore,
114        signer: Arc<dyn DurableAnchorSigner>,
115        bearer_verifier: CrossCheckedBearerVerifier,
116        clock: Clock,
117    ) -> Self {
118        Self {
119            store,
120            signer,
121            bearer_verifier,
122            clock,
123        }
124    }
125}
126
127/// Parsed runtime configuration. All fields are required together; an all-
128/// absent configuration keeps the route disabled.
129#[derive(Clone, PartialEq, Eq)]
130pub struct CrossCheckedAnchorStartupConfig {
131    bearer_verifier: CrossCheckedBearerVerifier,
132    expected_audience: String,
133    intermediate_did: String,
134    intermediate_key_id: String,
135    intermediate_public_key: PublicKey,
136    governance_frost_group_public_key: [u8; 32],
137    governance_frost_key_epoch: u64,
138    node_key_id: String,
139}
140
141impl fmt::Debug for CrossCheckedAnchorStartupConfig {
142    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
143        formatter
144            .debug_struct("CrossCheckedAnchorStartupConfig")
145            .field("bearer_verifier", &"[REDACTED]")
146            .field("expected_audience", &self.expected_audience)
147            .field("intermediate_did", &self.intermediate_did)
148            .field("intermediate_key_id", &self.intermediate_key_id)
149            .field("intermediate_public_key", &self.intermediate_public_key)
150            .field(
151                "governance_frost_key_epoch",
152                &self.governance_frost_key_epoch,
153            )
154            .field("node_key_id", &self.node_key_id)
155            .finish()
156    }
157}
158
159impl CrossCheckedAnchorStartupConfig {
160    /// Parse a complete environment snapshot without reading process-global
161    /// environment state, making startup policy deterministic and testable.
162    ///
163    /// # Errors
164    /// Fails closed if any field is missing, malformed, or reuses the admin
165    /// bearer.
166    pub fn from_values(
167        admin_bearer: &str,
168        values: &BTreeMap<String, String>,
169    ) -> Result<Option<Self>, CrossCheckedAnchorStartupError> {
170        let names = required_environment_names();
171        let present = names
172            .iter()
173            .filter(|name| values.contains_key(**name))
174            .count();
175        if present == 0 {
176            return Ok(None);
177        }
178        if present != names.len() {
179            return Err(CrossCheckedAnchorStartupError::IncompleteConfiguration);
180        }
181
182        let value = |name: &str| {
183            values
184                .get(name)
185                .cloned()
186                .ok_or(CrossCheckedAnchorStartupError::IncompleteConfiguration)
187        };
188        let bearer = Zeroizing::new(value(CROSSCHECKED_ANCHOR_BEARER_ENV)?);
189        let bearer_verifier = CrossCheckedBearerVerifier::from_bearer(&bearer)
190            .ok_or(CrossCheckedAnchorStartupError::InvalidBearer)?;
191        if constant_time_eq(bearer.as_bytes(), admin_bearer.as_bytes()) {
192            return Err(CrossCheckedAnchorStartupError::AdminBearerReuse);
193        }
194
195        let expected_audience = value(CROSSCHECKED_ANCHOR_EXPECTED_AUDIENCE_ENV)?;
196        if !is_closed_identifier(&expected_audience, 128) {
197            return Err(CrossCheckedAnchorStartupError::InvalidField("audience"));
198        }
199        let intermediate_did = value(CROSSCHECKED_ANCHOR_INTERMEDIATE_DID_ENV)?;
200        let intermediate_key_id = value(CROSSCHECKED_ANCHOR_INTERMEDIATE_KEY_ID_ENV)?;
201        validate_key_id(
202            &intermediate_did,
203            &intermediate_key_id,
204            "intermediate_key_id",
205        )?;
206        let public_key_hex = value(CROSSCHECKED_ANCHOR_INTERMEDIATE_PUBLIC_KEY_ENV)?;
207        let intermediate_public_key = PublicKey::from_bytes(decode_exact_hex::<32>(
208            &public_key_hex,
209            "intermediate_public_key",
210        )?);
211        let governance_frost_group_public_key = decode_exact_hex::<32>(
212            &value(CROSSCHECKED_ANCHOR_GOVERNANCE_FROST_PUBLIC_KEY_ENV)?,
213            "governance_frost_group_public_key",
214        )?;
215        frost_ed25519::VerifyingKey::deserialize(&governance_frost_group_public_key).map_err(
216            |_| CrossCheckedAnchorStartupError::InvalidField("governance_frost_group_public_key"),
217        )?;
218        let governance_frost_key_epoch = value(CROSSCHECKED_ANCHOR_GOVERNANCE_FROST_KEY_EPOCH_ENV)?
219            .parse::<u64>()
220            .map_err(|_| {
221                CrossCheckedAnchorStartupError::InvalidField("governance_frost_key_epoch")
222            })?;
223        if governance_frost_key_epoch == 0 {
224            return Err(CrossCheckedAnchorStartupError::InvalidField(
225                "governance_frost_key_epoch",
226            ));
227        }
228        let node_key_id = value(CROSSCHECKED_ANCHOR_NODE_KEY_ID_ENV)?;
229        if node_key_id.is_empty() || node_key_id.len() > 321 {
230            return Err(CrossCheckedAnchorStartupError::InvalidField("node_key_id"));
231        }
232
233        Ok(Some(Self {
234            bearer_verifier,
235            expected_audience,
236            intermediate_did,
237            intermediate_key_id,
238            intermediate_public_key,
239            governance_frost_group_public_key,
240            governance_frost_key_epoch,
241            node_key_id,
242        }))
243    }
244
245    #[must_use]
246    pub fn bearer_verifier(&self) -> CrossCheckedBearerVerifier {
247        self.bearer_verifier.clone()
248    }
249
250    #[must_use]
251    pub fn node_key_id(&self) -> &str {
252        &self.node_key_id
253    }
254
255    #[must_use]
256    pub fn store_config(&self, node_did: String, node_public_key: PublicKey) -> AnchorStoreConfig {
257        AnchorStoreConfig {
258            expected_audience: self.expected_audience.clone(),
259            crosschecked_intermediate_did: self.intermediate_did.clone(),
260            crosschecked_intermediate_key_id: self.intermediate_key_id.clone(),
261            crosschecked_intermediate_public_key: self.intermediate_public_key,
262            governance_frost_group_public_key: self.governance_frost_group_public_key,
263            governance_frost_key_epoch: self.governance_frost_key_epoch,
264            node_identity: AnchorNodeIdentity {
265                did: node_did,
266                key_id: self.node_key_id.clone(),
267                public_key: node_public_key,
268            },
269        }
270    }
271}
272
273#[derive(Clone, Debug, Error, PartialEq, Eq)]
274pub enum CrossCheckedAnchorStartupError {
275    #[error("CrossChecked anchor configuration is incomplete")]
276    IncompleteConfiguration,
277    #[error("CrossChecked anchor bearer must be exactly 256-bit lowercase hex")]
278    InvalidBearer,
279    #[error("CrossChecked anchor bearer must not reuse the admin bearer")]
280    AdminBearerReuse,
281    #[error("invalid CrossChecked anchor configuration field: {0}")]
282    InvalidField(&'static str),
283}
284
285/// Build the exact commitment-only POST and authenticated readback routes.
286pub fn crosschecked_anchor_router(state: CrossCheckedAnchorHttpState) -> Router {
287    Router::new()
288        .route(ANCHOR_PATH, post(post_anchor))
289        .route(&format!("{ANCHOR_PATH}/:action_hash"), get(readback_anchor))
290        .layer(DefaultBodyLimit::max(MAX_REQUEST_BODY_BYTES))
291        .with_state(state)
292}
293
294async fn post_anchor(
295    State(state): State<CrossCheckedAnchorHttpState>,
296    headers: HeaderMap,
297    body: Bytes,
298) -> Response {
299    if let Err(status) = authorize(&headers, &state.bearer_verifier) {
300        return status.into_response();
301    }
302    if single_header(&headers, header::CONTENT_TYPE) != Some("application/cbor") {
303        return error_response(StatusCode::UNSUPPORTED_MEDIA_TYPE, "invalid_content_type");
304    }
305
306    let store = state.store.clone();
307    let signer = Arc::clone(&state.signer);
308    let clock = Arc::clone(&state.clock);
309    let body = body.to_vec();
310    let result = tokio::task::spawn_blocking(move || {
311        let now = clock().map_err(|_| AnchorStoreError::Storage("clock unavailable".into()))?;
312        store.record(
313            SubmissionContext {
314                method: "POST",
315                path: ANCHOR_PATH,
316                content_type: "application/cbor",
317                body: &body,
318                now,
319            },
320            signer.as_ref(),
321        )
322    })
323    .await;
324
325    match result {
326        Ok(Ok(record)) => {
327            let status = match record.disposition {
328                AnchorRecordDisposition::Created => StatusCode::CREATED,
329                AnchorRecordDisposition::Replayed => StatusCode::OK,
330            };
331            (
332                status,
333                [(header::CONTENT_TYPE, "application/cbor")],
334                record.response_body,
335            )
336                .into_response()
337        }
338        Ok(Err(error)) => store_error_response(error),
339        Err(_) => error_response(StatusCode::SERVICE_UNAVAILABLE, "worker_unavailable"),
340    }
341}
342
343async fn readback_anchor(
344    State(state): State<CrossCheckedAnchorHttpState>,
345    AxumPath(action_hash): AxumPath<String>,
346    headers: HeaderMap,
347) -> Response {
348    if let Err(status) = authorize(&headers, &state.bearer_verifier) {
349        return status.into_response();
350    }
351    let action_hash = match decode_exact_hex::<32>(&action_hash, "action_hash") {
352        Ok(action_hash) => action_hash,
353        Err(_) => return error_response(StatusCode::BAD_REQUEST, "invalid_action_hash"),
354    };
355    let store = state.store.clone();
356    match tokio::task::spawn_blocking(move || store.readback_action(SOURCE_CODE, &action_hash))
357        .await
358    {
359        Ok(Ok(Some(readback))) => (
360            StatusCode::OK,
361            [(header::CONTENT_TYPE, "application/cbor")],
362            readback.response_body,
363        )
364            .into_response(),
365        Ok(Ok(None)) => error_response(StatusCode::NOT_FOUND, "anchor_not_found"),
366        Ok(Err(_)) | Err(_) => {
367            error_response(StatusCode::SERVICE_UNAVAILABLE, "readback_unavailable")
368        }
369    }
370}
371
372fn authorize(headers: &HeaderMap, verifier: &CrossCheckedBearerVerifier) -> Result<(), StatusCode> {
373    let Some(value) = single_header(headers, header::AUTHORIZATION) else {
374        return Err(StatusCode::UNAUTHORIZED);
375    };
376    let Some(provided) = value.strip_prefix("Bearer ") else {
377        return Err(StatusCode::UNAUTHORIZED);
378    };
379    if verifier.verifies(provided) {
380        Ok(())
381    } else {
382        Err(StatusCode::FORBIDDEN)
383    }
384}
385
386fn single_header(headers: &HeaderMap, name: header::HeaderName) -> Option<&str> {
387    let mut values = headers.get_all(name).iter();
388    let value = values.next()?;
389    if values.next().is_some() {
390        return None;
391    }
392    value.to_str().ok()
393}
394
395fn store_error_response(error: AnchorStoreError) -> Response {
396    match error {
397        AnchorStoreError::Conflict(AnchorConflictKind::IdempotencyKey) => {
398            error_response(StatusCode::CONFLICT, "idempotency_key_conflict")
399        }
400        AnchorStoreError::Conflict(AnchorConflictKind::ActionHash) => {
401            error_response(StatusCode::CONFLICT, "action_hash_conflict")
402        }
403        AnchorStoreError::Codec(_) => error_response(StatusCode::BAD_REQUEST, "invalid_request"),
404        AnchorStoreError::AuthorityValidation(_)
405        | AnchorStoreError::AuthorityNotFound
406        | AnchorStoreError::AuthorityNotYetValid
407        | AnchorStoreError::AuthorityExpired
408        | AnchorStoreError::AuthorityRetired => {
409            error_response(StatusCode::FORBIDDEN, "authority_rejected")
410        }
411        AnchorStoreError::Signer(_)
412        | AnchorStoreError::Storage(_)
413        | AnchorStoreError::GovernanceAuthorization(_)
414        | AnchorStoreError::GovernanceAuthorizationConflict
415        | AnchorStoreError::ReadbackValidation(_) => {
416            error_response(StatusCode::SERVICE_UNAVAILABLE, "recording_unavailable")
417        }
418    }
419}
420
421fn error_response(status: StatusCode, error_code: &'static str) -> Response {
422    (status, Json(json!({ "error_code": error_code }))).into_response()
423}
424
425/// SQLite signature journal binding each operation to one payload and one
426/// signature across process restarts.
427pub struct SqliteDurableAnchorSigner {
428    path: PathBuf,
429    identity: AnchorNodeIdentity,
430    sign: SignFunction,
431}
432
433impl fmt::Debug for SqliteDurableAnchorSigner {
434    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
435        formatter
436            .debug_struct("SqliteDurableAnchorSigner")
437            .field("path", &self.path)
438            .field("identity", &self.identity)
439            .field("sign", &"[NON-EXPORTABLE CALLBACK]")
440            .finish()
441    }
442}
443
444impl SqliteDurableAnchorSigner {
445    /// Open or create a durable signature journal pinned to one node identity.
446    ///
447    /// # Errors
448    /// Fails closed on inaccessible storage or identity-policy mismatch.
449    pub fn open(
450        path: impl AsRef<Path>,
451        identity: AnchorNodeIdentity,
452        sign: SignFunction,
453    ) -> Result<Self, SignOnceError> {
454        let signer = Self {
455            path: path.as_ref().to_path_buf(),
456            identity,
457            sign,
458        };
459        signer.initialize()?;
460        Ok(signer)
461    }
462
463    fn connection(&self) -> Result<Connection, SignOnceError> {
464        let connection = Connection::open(&self.path).map_err(signer_storage)?;
465        connection
466            .busy_timeout(std::time::Duration::from_secs(30))
467            .map_err(signer_storage)?;
468        connection
469            .execute_batch("PRAGMA synchronous = FULL;")
470            .map_err(signer_storage)?;
471        Ok(connection)
472    }
473
474    fn initialize(&self) -> Result<(), SignOnceError> {
475        let mut connection = self.connection()?;
476        let transaction = connection
477            .transaction_with_behavior(TransactionBehavior::Immediate)
478            .map_err(signer_storage)?;
479        transaction
480            .execute_batch(
481                "CREATE TABLE IF NOT EXISTS crosschecked_anchor_signer_config (
482                    singleton INTEGER PRIMARY KEY CHECK (singleton = 1),
483                    schema_version TEXT NOT NULL,
484                    node_did TEXT NOT NULL,
485                    node_key_id TEXT NOT NULL,
486                    node_public_key BLOB NOT NULL
487                 );
488                 CREATE TABLE IF NOT EXISTS crosschecked_anchor_signatures (
489                    operation_id BLOB PRIMARY KEY,
490                    payload_hash BLOB NOT NULL,
491                    signature BLOB NOT NULL
492                 );
493                 CREATE TABLE IF NOT EXISTS crosschecked_anchor_recorded_at_reservations (
494                    request_hash BLOB PRIMARY KEY CHECK (length(request_hash) = 32),
495                    node_recorded_physical_ms INTEGER NOT NULL
496                        CHECK (node_recorded_physical_ms >= 0),
497                    node_recorded_logical INTEGER NOT NULL
498                        CHECK (node_recorded_logical >= 0)
499                 );",
500            )
501            .map_err(signer_storage)?;
502        transaction
503            .execute(
504                "INSERT OR IGNORE INTO crosschecked_anchor_signer_config
505                 (singleton, schema_version, node_did, node_key_id, node_public_key)
506                 VALUES (1, ?1, ?2, ?3, ?4)",
507                params![
508                    SIGNER_SCHEMA_VERSION,
509                    self.identity.did,
510                    self.identity.key_id,
511                    self.identity.public_key.as_bytes().as_slice(),
512                ],
513            )
514            .map_err(signer_storage)?;
515        let stored: (String, String, String, Vec<u8>) = transaction
516            .query_row(
517                "SELECT schema_version, node_did, node_key_id, node_public_key
518                 FROM crosschecked_anchor_signer_config WHERE singleton = 1",
519                [],
520                |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)),
521            )
522            .map_err(signer_storage)?;
523        if stored.0 != SIGNER_SCHEMA_VERSION
524            || stored.1 != self.identity.did
525            || stored.2 != self.identity.key_id
526            || stored.3 != self.identity.public_key.as_bytes()
527        {
528            return Err(SignOnceError::Unavailable(
529                "persistent signer identity policy mismatch".into(),
530            ));
531        }
532        transaction.commit().map_err(signer_storage)
533    }
534}
535
536impl DurableAnchorSigner for SqliteDurableAnchorSigner {
537    fn identity(&self) -> AnchorNodeIdentity {
538        self.identity.clone()
539    }
540
541    fn reserved_recorded_at(
542        &self,
543        request_hash: Hash256,
544    ) -> Result<Option<Timestamp>, SignOnceError> {
545        let connection = self.connection()?;
546        connection
547            .query_row(
548                "SELECT node_recorded_physical_ms, node_recorded_logical
549                 FROM crosschecked_anchor_recorded_at_reservations
550                 WHERE request_hash = ?1",
551                [request_hash.as_bytes().as_slice()],
552                |row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?)),
553            )
554            .optional()
555            .map_err(signer_storage)?
556            .map(|(physical_ms, logical)| signer_timestamp(physical_ms, logical))
557            .transpose()
558    }
559
560    fn reserve_recorded_at(
561        &self,
562        request_hash: Hash256,
563        proposed: Timestamp,
564    ) -> Result<Timestamp, SignOnceError> {
565        let physical_ms = i64::try_from(proposed.physical_ms).map_err(|_| {
566            SignOnceError::Unavailable(
567                "node recorded physical timestamp exceeds SQLite INTEGER".into(),
568            )
569        })?;
570        let logical = i64::from(proposed.logical);
571        let mut connection = self.connection()?;
572        let transaction = connection
573            .transaction_with_behavior(TransactionBehavior::Immediate)
574            .map_err(signer_storage)?;
575        transaction
576            .execute(
577                "INSERT OR IGNORE INTO crosschecked_anchor_recorded_at_reservations
578                 (request_hash, node_recorded_physical_ms, node_recorded_logical)
579                 VALUES (?1, ?2, ?3)",
580                params![request_hash.as_bytes().as_slice(), physical_ms, logical],
581            )
582            .map_err(signer_storage)?;
583        let stored = transaction
584            .query_row(
585                "SELECT node_recorded_physical_ms, node_recorded_logical
586                 FROM crosschecked_anchor_recorded_at_reservations
587                 WHERE request_hash = ?1",
588                [request_hash.as_bytes().as_slice()],
589                |row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?)),
590            )
591            .map_err(signer_storage)?;
592        let reserved = signer_timestamp(stored.0, stored.1)?;
593        transaction.commit().map_err(signer_storage)?;
594        Ok(reserved)
595    }
596
597    fn sign_once(&self, operation_id: Hash256, payload: &[u8]) -> Result<Signature, SignOnceError> {
598        let payload_hash = Hash256::digest(payload);
599        let mut connection = self.connection()?;
600        let transaction = connection
601            .transaction_with_behavior(TransactionBehavior::Immediate)
602            .map_err(signer_storage)?;
603        let existing: Option<(Vec<u8>, Vec<u8>)> = transaction
604            .query_row(
605                "SELECT payload_hash, signature FROM crosschecked_anchor_signatures
606                 WHERE operation_id = ?1",
607                [operation_id.as_bytes().as_slice()],
608                |row| Ok((row.get(0)?, row.get(1)?)),
609            )
610            .optional()
611            .map_err(signer_storage)?;
612        if let Some((stored_hash, stored_signature)) = existing {
613            if stored_hash != payload_hash.as_bytes() {
614                return Err(SignOnceError::OperationPayloadConflict);
615            }
616            let signature = exact_array::<64>(stored_signature, "stored signature")?;
617            transaction.commit().map_err(signer_storage)?;
618            return Ok(Signature::from_bytes(signature));
619        }
620
621        let signature = (self.sign)(payload);
622        let signature_bytes = signature.ed25519_bytes().ok_or_else(|| {
623            SignOnceError::Unavailable("node signer did not return an Ed25519 signature".into())
624        })?;
625        transaction
626            .execute(
627                "INSERT INTO crosschecked_anchor_signatures
628                 (operation_id, payload_hash, signature) VALUES (?1, ?2, ?3)",
629                params![
630                    operation_id.as_bytes().as_slice(),
631                    payload_hash.as_bytes().as_slice(),
632                    signature_bytes.as_slice(),
633                ],
634            )
635            .map_err(signer_storage)?;
636        transaction.commit().map_err(signer_storage)?;
637        Ok(signature)
638    }
639}
640
641fn required_environment_names() -> [&'static str; 8] {
642    [
643        CROSSCHECKED_ANCHOR_BEARER_ENV,
644        CROSSCHECKED_ANCHOR_EXPECTED_AUDIENCE_ENV,
645        CROSSCHECKED_ANCHOR_INTERMEDIATE_DID_ENV,
646        CROSSCHECKED_ANCHOR_INTERMEDIATE_KEY_ID_ENV,
647        CROSSCHECKED_ANCHOR_INTERMEDIATE_PUBLIC_KEY_ENV,
648        CROSSCHECKED_ANCHOR_GOVERNANCE_FROST_PUBLIC_KEY_ENV,
649        CROSSCHECKED_ANCHOR_GOVERNANCE_FROST_KEY_EPOCH_ENV,
650        CROSSCHECKED_ANCHOR_NODE_KEY_ID_ENV,
651    ]
652}
653
654fn is_valid_bearer(bearer: &str) -> bool {
655    bearer.len() == 64
656        && bearer
657            .bytes()
658            .all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f'))
659}
660
661fn validate_key_id(
662    did: &str,
663    key_id: &str,
664    field: &'static str,
665) -> Result<(), CrossCheckedAnchorStartupError> {
666    let prefix = format!("{did}#");
667    let fragment = key_id
668        .strip_prefix(&prefix)
669        .ok_or(CrossCheckedAnchorStartupError::InvalidField(field))?;
670    if did.is_empty()
671        || did.len() > 256
672        || fragment.is_empty()
673        || fragment.len() > 64
674        || !fragment
675            .bytes()
676            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-'))
677    {
678        return Err(CrossCheckedAnchorStartupError::InvalidField(field));
679    }
680    Ok(())
681}
682
683fn is_closed_identifier(value: &str, max: usize) -> bool {
684    !value.is_empty()
685        && value.len() <= max
686        && value.bytes().all(|byte| {
687            byte.is_ascii_lowercase()
688                || byte.is_ascii_digit()
689                || matches!(byte, b'.' | b'_' | b':' | b'-')
690        })
691}
692
693fn decode_exact_hex<const N: usize>(
694    value: &str,
695    field: &'static str,
696) -> Result<[u8; N], CrossCheckedAnchorStartupError> {
697    if value.len() != N * 2
698        || !value
699            .bytes()
700            .all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f'))
701    {
702        return Err(CrossCheckedAnchorStartupError::InvalidField(field));
703    }
704    let mut bytes = [0_u8; N];
705    hex::decode_to_slice(value, &mut bytes)
706        .map_err(|_| CrossCheckedAnchorStartupError::InvalidField(field))?;
707    Ok(bytes)
708}
709
710fn exact_array<const N: usize>(bytes: Vec<u8>, field: &str) -> Result<[u8; N], SignOnceError> {
711    bytes.try_into().map_err(|_| {
712        SignOnceError::Unavailable(format!("{field} has an invalid persistent byte length"))
713    })
714}
715
716fn signer_storage(error: rusqlite::Error) -> SignOnceError {
717    SignOnceError::Unavailable(error.to_string())
718}
719
720fn signer_timestamp(physical_ms: i64, logical: i64) -> Result<Timestamp, SignOnceError> {
721    let physical_ms = u64::try_from(physical_ms).map_err(|_| {
722        SignOnceError::Unavailable("stored node recorded physical timestamp is negative".into())
723    })?;
724    let logical = u32::try_from(logical).map_err(|_| {
725        SignOnceError::Unavailable("stored node recorded logical timestamp is invalid".into())
726    })?;
727    Ok(Timestamp::new(physical_ms, logical))
728}
729
730fn constant_time_eq(left: &[u8], right: &[u8]) -> bool {
731    let mut difference = left.len() ^ right.len();
732    let length = left.len().max(right.len());
733    for index in 0..length {
734        let left_byte = left.get(index).copied().unwrap_or(0);
735        let right_byte = right.get(index).copied().unwrap_or(0);
736        difference |= usize::from(left_byte ^ right_byte);
737    }
738    difference == 0
739}
exo_node/crosschecked_anchor_http.rs

exo_node/
crosschecked_anchor_http.rs
