PDF Tamper Detection in Rust: Integration Guide

[dependencies]
reqwest = { version = "0.13", features = ["json"] }
tokio = { version = "1", features = ["macros", "rt-multi-thread", "net", "time"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
thiserror = "2"
uuid = { version = "1", features = ["v4"] }

# Step 5 (HTTP handler) and Step 6 (presigned URLs) only
axum = "0.8"
aws-config = "1"
aws-sdk-s3 = "1"

// src/htpbe/mod.rs
pub mod client;
pub mod error;
pub mod routing;
pub mod types;

curl -i -X POST https://api.htpbe.tech/v1/analyze \
  -H "Authorization: Bearer htpbe_test_YOUR_TEST_KEY" \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: 7c2f6b6e-1a3d-4b8a-9c3a-2e6f1d9b4a10" \
  -d '{"url": "https://api.htpbe.tech/v1/test/signature-removed.pdf"}'

{ "id": "00000000-0000-4000-8000-00000000000c" }

curl https://api.htpbe.tech/v1/result/00000000-0000-4000-8000-00000000000c \
  -H "Authorization: Bearer htpbe_test_YOUR_TEST_KEY"

{
  "id": "00000000-0000-4000-8000-00000000000c",
  "filename": "signature-removed.pdf",
  "check_date": null,
  "status": "modified",
  "modification_confidence": "certain",
  "modification_markers": ["HTPBE_SIGNATURE_REMOVED"],
  "signature_removed": true,
  "modifications_after_signature": false,
  "usage_caution": {
    "safe_for_automated_adverse_decision": false,
    "recommended_action": "route_to_human_review",
    "message": "Structural evidence of post-creation modification was found. Route this document to a human reviewer before any decision — do not use this verdict as the sole basis for an automatic adverse decision against a person."
  }
}

// src/htpbe/types.rs
use serde::{Deserialize, Serialize};

#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Status {
    Intact,
    Modified,
    Inconclusive,
    #[serde(other)]
    Unknown,
}

/// Present only when `status` is `inconclusive`. It explains why the
/// integrity check could not reach `intact` or `modified`.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum StatusReason {
    ConsumerSoftwareOrigin,
    OnlineEditorOrigin,
    ScannedDocument,
    UnverifiableMetadata,
    FilledFormOrigin,
    FillSignOrigin,
    VisibleSignatureOrigin,
    /// Describes the check, not the document: part of the analysis did not finish.
    AnalysisIncomplete,
    #[serde(other)]
    Unknown,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Confidence {
    Certain,
    High,
    /// The wire value is "none". Renamed so it doesn't read like `Option::None`.
    #[serde(rename = "none")]
    NoneDetected,
    #[serde(other)]
    Unknown,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum RecommendedAction {
    RouteToHumanReview,
    RequestFromIssuer,
    NoAction,
    #[serde(other)]
    Unknown,
}

#[derive(Debug, Clone, Deserialize)]
pub struct UsageCaution {
    /// Always false. The verdict alone must never drive an adverse decision.
    pub safe_for_automated_adverse_decision: bool,
    pub recommended_action: RecommendedAction,
    pub message: String,
}

/// The flat GET /result/{id} response. Partial on purpose: serde skips
/// any field not declared here, so new API fields never break decoding.
#[derive(Debug, Clone, Deserialize)]
pub struct CheckResult {
    pub id: String,
    pub filename: String,
    /// Unix seconds. Null for test-key results and legacy records.
    pub check_date: Option<i64>,
    pub file_size: u64,

    pub algorithm_version: String,
    pub current_algorithm_version: String,
    /// Present only when the check ran on an older algorithm version.
    pub outdated_warning: Option<String>,

    pub status: Status,
    pub status_reason: Option<StatusReason>,
    /// Null on legacy records.
    pub modification_confidence: Option<Confidence>,
    /// Public HTPBE_* ids, strongest first.
    pub modification_markers: Vec<String>,

    pub has_digital_signature: bool,
    pub signature_removed: bool,
    pub modifications_after_signature: bool,

    pub page_count: u32,

    pub usage_caution: UsageCaution,
}

// src/htpbe/error.rs
use std::time::Duration;

#[derive(Debug, thiserror::Error)]
pub enum HtpbeError {
    /// A non-2xx response from the API (or from something in front of it).
    #[error("HTPBE API returned {status} ({code}): {message}")]
    Api {
        status: u16,
        /// Machine-readable code, e.g. "payment_required". "unknown" when
        /// the body was not the API's JSON error shape.
        code: String,
        message: String,
        details: Option<String>,
        /// Parsed from Retry-After (seconds) when the response carried it.
        retry_after: Option<Duration>,
    },

    /// Connect failure, timeout, reset connection, unreadable body.
    #[error("transport error: {0}")]
    Transport(#[from] reqwest::Error),

    /// A 2xx response whose body did not match the documented shape.
    #[error("unexpected response body: {0}")]
    Decode(#[from] serde_json::Error),

    #[error("configuration error: {0}")]
    Config(&'static str),
}

impl HtpbeError {
    /// Whether sending the SAME request again (same Idempotency-Key) could succeed.
    pub fn is_retryable(&self) -> bool {
        match self {
            // 409 idempotency_key_in_progress: the original is still running.
            // 429 server_at_capacity: server-wide concurrency, not your quota.
            // 500/502/504: transient on the server side or in front of it.
            // 503 is NOT retried here: it surfaces to the caller (CheckPending).
            HtpbeError::Api { status, .. } => matches!(*status, 409 | 429 | 500 | 502 | 504),
            // Timeouts and dropped connections are exactly what the
            // Idempotency-Key exists for. A builder error is a bug, not a blip.
            HtpbeError::Transport(e) => !e.is_builder(),
            // A malformed 2xx body will be malformed again.
            HtpbeError::Decode(_) | HtpbeError::Config(_) => false,
        }
    }

    pub fn code(&self) -> Option<&str> {
        match self {
            HtpbeError::Api { code, .. } => Some(code.as_str()),
            _ => None,
        }
    }

    pub fn retry_after(&self) -> Option<Duration> {
        match self {
            HtpbeError::Api { retry_after, .. } => *retry_after,
            _ => None,
        }
    }

    /// 409 idempotency_key_in_progress: the original request with this key
    /// is still running. The retry loop waits on it separately.
    pub fn is_in_progress(&self) -> bool {
        matches!(self, HtpbeError::Api { status: 409, .. })
    }

    /// No credit source left on the account. Not retryable: it fails the
    /// same way until credits are added, so send it to whoever owns billing.
    pub fn is_out_of_credits(&self) -> bool {
        self.code() == Some("payment_required")
    }
}

// src/htpbe/client.rs
use std::future::Future;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use reqwest::header::RETRY_AFTER;
use reqwest::Response;
use serde::{Deserialize, Serialize};

use super::error::HtpbeError;
use super::types::CheckResult;

const DEFAULT_BASE_URL: &str = "https://api.htpbe.tech/v1";
const MAX_RETRY_AFTER_SECS: u64 = 60;
/// How long to keep waiting on a 409 (original request still running).
const MAX_IN_PROGRESS_WAIT: Duration = Duration::from_secs(90);

#[derive(Clone)]
pub struct HtpbeClient {
    http: reqwest::Client,
    base_url: String,
    api_key: String,
    max_attempts: u32,
}

#[derive(Serialize)]
struct AnalyzeRequest<'a> {
    url: &'a str,
    #[serde(skip_serializing_if = "Option::is_none")]
    original_filename: Option<&'a str>,
}

#[derive(Deserialize)]
struct AnalyzeResponse {
    id: String,
}

#[derive(Deserialize)]
struct ErrorBody {
    error: String,
    code: String,
    details: Option<String>,
}

impl HtpbeClient {
    pub fn new(api_key: impl Into<String>) -> Result<Self, HtpbeError> {
        let api_key = api_key.into();
        if api_key.is_empty() {
            return Err(HtpbeError::Config("HTPBE API key is empty"));
        }
        let http = reqwest::Client::builder()
            .connect_timeout(Duration::from_secs(10))
            // Above the server's 30 s download + 20 s analysis limits,
            // so a slow but valid check is never cut off client-side.
            .timeout(Duration::from_secs(90))
            .build()?;
        Ok(Self {
            http,
            base_url: DEFAULT_BASE_URL.to_owned(),
            api_key,
            max_attempts: 4,
        })
    }

    /// Point the client at a mock server in tests.
    #[allow(dead_code)]
    pub fn with_base_url(mut self, base_url: impl Into<String>) -> Self {
        let base_url: String = base_url.into();
        self.base_url = base_url.trim_end_matches('/').to_owned();
        self
    }

    /// One logical check: submit (retried under the caller's key), then read.
    /// Persist `idempotency_key` with the case and reuse it if you retry later.
    pub async fn check_pdf(
        &self,
        pdf_url: &str,
        original_filename: Option<&str>,
        idempotency_key: &str,
    ) -> Result<CheckResult, HtpbeError> {
        let id = self.analyze(pdf_url, original_filename, idempotency_key).await?;
        self.result(&id).await
    }

    /// POST /analyze. Pass the same `idempotency_key` on every retry of
    /// one logical request, and a new key for every new document.
    pub async fn analyze(
        &self,
        pdf_url: &str,
        original_filename: Option<&str>,
        idempotency_key: &str,
    ) -> Result<String, HtpbeError> {
        let body = AnalyzeRequest { url: pdf_url, original_filename };
        let body = &body;

        self.with_retry(|| async move {
            let resp = self
                .http
                .post(format!("{}/analyze", self.base_url))
                .bearer_auth(&self.api_key)
                .header("Idempotency-Key", idempotency_key)
                .json(body)
                .send()
                .await?;
            let resp = check_status(resp).await?;
            let bytes = resp.bytes().await?;
            let parsed: AnalyzeResponse = serde_json::from_slice(&bytes)?;
            Ok(parsed.id)
        })
        .await
    }

    /// GET /result/{id}. A plain read, so retrying it is always safe.
    pub async fn result(&self, id: &str) -> Result<CheckResult, HtpbeError> {
        self.with_retry(|| async move {
            let resp = self
                .http
                .get(format!("{}/result/{}", self.base_url, id))
                .bearer_auth(&self.api_key)
                .send()
                .await?;
            let resp = check_status(resp).await?;
            let bytes = resp.bytes().await?;
            Ok(serde_json::from_slice(&bytes)?)
        })
        .await
    }

    async fn with_retry<T, F, Fut>(&self, mut op: F) -> Result<T, HtpbeError>
    where
        F: FnMut() -> Fut,
        Fut: Future<Output = Result<T, HtpbeError>>,
    {
        let mut retries: u32 = 0;
        let mut waited_in_progress = Duration::ZERO;
        loop {
            match op().await {
                Ok(value) => return Ok(value),
                // The original request is still running: wait on it without
                // spending a normal attempt, up to MAX_IN_PROGRESS_WAIT.
                Err(err) if err.is_in_progress() && waited_in_progress < MAX_IN_PROGRESS_WAIT => {
                    let delay = err
                        .retry_after()
                        .unwrap_or(Duration::from_secs(2))
                        .min(MAX_IN_PROGRESS_WAIT - waited_in_progress);
                    waited_in_progress += delay;
                    tokio::time::sleep(delay).await;
                }
                // 409 is excluded here, so the in-progress budget is a hard stop.
                Err(err) if !err.is_in_progress() && err.is_retryable() && retries + 1 < self.max_attempts => {
                    retries += 1;
                    let delay = err.retry_after().unwrap_or_else(|| backoff(retries));
                    tokio::time::sleep(delay).await;
                }
                Err(err) => return Err(err),
            }
        }
    }
}

/// Turns a non-2xx response into HtpbeError::Api. The body is parsed on a
/// best-effort basis: a proxy in front of the API can answer with HTML.
async fn check_status(resp: Response) -> Result<Response, HtpbeError> {
    let status = resp.status();
    if status.is_success() {
        return Ok(resp);
    }

    let retry_after = resp
        .headers()
        .get(RETRY_AFTER)
        .and_then(|v| v.to_str().ok())
        .and_then(|s| s.trim().parse::<u64>().ok())
        .map(|secs| Duration::from_secs(secs.clamp(1, MAX_RETRY_AFTER_SECS)));

    let bytes = resp.bytes().await.unwrap_or_default();
    let (code, message, details) = match serde_json::from_slice::<ErrorBody>(&bytes) {
        Ok(body) => (body.code, body.error, body.details),
        Err(_) => (
            "unknown".to_owned(),
            status.canonical_reason().unwrap_or("unexpected status").to_owned(),
            None,
        ),
    };

    Err(HtpbeError::Api {
        status: status.as_u16(),
        code,
        message,
        details,
        retry_after,
    })
}

/// 1 s, 2 s, 4 s ... capped at 30 s, plus up to one second of jitter.
/// The jitter comes from the clock's sub-second nanos: enough to keep a
/// batch of workers from retrying in lockstep, without an RNG crate.
fn backoff(attempt: u32) -> Duration {
    let base = Duration::from_secs(1u64 << attempt.saturating_sub(1).min(5))
        .min(Duration::from_secs(30));
    let nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.subsec_nanos())
        .unwrap_or(0);
    base + Duration::from_millis(u64::from(nanos % 1_000))
}

// src/htpbe/routing.rs
use super::types::{CheckResult, Confidence, RecommendedAction, Status, StatusReason};

/// Where a checked document goes next.
///
/// IMPORTANT: no variant here declines an application, flags a person or
/// takes any adverse action. The API marks every verdict
/// `safe_for_automated_adverse_decision: false`. A structural verdict is a
/// statement about the FILE, never about the person who sent it.
#[derive(Debug)]
pub enum Route {
    /// intact + no_action: no structural modification was detected.
    /// Not a statement that the document is authentic or its contents true.
    Proceed,
    /// modified + route_to_human_review, and every unknown or inconsistent case.
    HumanReview {
        markers: Vec<String>,
        confidence: Option<Confidence>,
    },
    /// inconclusive + request_from_issuer: the limit of what the file's
    /// structure can show. Not a suspicion of fraud.
    RequestFromIssuer { reason: Option<StatusReason> },
    /// status_reason analysis_incomplete: run the check again with a NEW
    /// Idempotency-Key. If it repeats, the caller downgrades to RequestFromIssuer.
    Resubmit,
}

pub fn route_check(r: &CheckResult) -> Route {
    // The incomplete case first: it describes the check, not the document.
    // (An incomplete analysis that still found a modification comes back as
    // `modified` with no status_reason, and the match below routes it.)
    if r.status_reason == Some(StatusReason::AnalysisIncomplete) {
        return Route::Resubmit;
    }

    match (r.usage_caution.recommended_action, r.status) {
        (RecommendedAction::NoAction, Status::Intact) => Route::Proceed,
        (RecommendedAction::RequestFromIssuer, Status::Inconclusive) => Route::RequestFromIssuer {
            reason: r.status_reason,
        },
        // route_to_human_review, plus anything unknown or inconsistent.
        // Fail safe: an unrecognised answer goes to a person, never to Proceed.
        _ => Route::HumanReview {
            markers: r.modification_markers.clone(),
            confidence: r.modification_confidence,
        },
    }
}

// src/main.rs
mod htpbe;

use axum::{extract::State, http::StatusCode, routing::post, Json, Router};
use serde::{Deserialize, Serialize};

use htpbe::client::HtpbeClient;
use htpbe::error::HtpbeError;
use htpbe::routing::{route_check, Route};
use htpbe::types::{Confidence, StatusReason};
use uuid::Uuid;

#[derive(Clone)]
struct AppState {
    htpbe: HtpbeClient,
}

#[derive(Deserialize)]
struct SubmitDocument {
    document_url: String,
    original_filename: Option<String>,
    case_id: String,
    /// Minted (UUID v4) when the case was recorded and stored with it, next to
    /// the exact document_url. A pending case is retried with the SAME pair.
    check_key: String,
    /// Set once a first run came back analysis_incomplete. Stored on the case.
    resubmit_key: Option<String>,
}

#[derive(Serialize)]
#[serde(tag = "route", rename_all = "snake_case")]
enum CaseOutcome {
    Proceed { check_id: String },
    HumanReview {
        check_id: String,
        markers: Vec<String>,
        confidence: Option<Confidence>,
    },
    RequestFromIssuer { check_id: String, reason: Option<StatusReason> },
    /// The file itself could not be analysed (too large, unreadable,
    /// password-protected). Ask the submitter for a readable copy.
    FileNotCheckable { error_code: String },
    /// The check did not happen. Not a pass, not a fail: the case waits.
    CheckPending,
}

/// Stand-in for your case store: write the resubmission key to the case
/// record. Replace with a real database write.
async fn save_resubmit_key(case_id: &str, key: &str) -> std::io::Result<()> {
    // Demo only: never log real idempotency keys.
    let _ = (case_id, key);
    Ok(())
}

/// Runs the check and, for analysis_incomplete, exactly one fresh run.
async fn run_check(client: &HtpbeClient, req: &SubmitDocument) -> Result<(String, Route), HtpbeError> {
    let url = req.document_url.as_str();
    let name = req.original_filename.as_deref();

    let resubmit_key = match &req.resubmit_key {
        // A stored resubmit key means the first run already came back
        // incomplete: replay the resubmission, don't repeat the first check.
        Some(key) => key.clone(),
        None => {
            let first = client.check_pdf(url, name, &req.check_key).await?;
            match route_check(&first) {
                Route::Resubmit => {
                    // A resubmission is a new logical request, so it gets a NEW
                    // Idempotency-Key, persisted on the case BEFORE it is sent.
                    // If the write fails, nothing is sent and the case stays pending.
                    let key = Uuid::new_v4().to_string();
                    save_resubmit_key(&req.case_id, &key)
                        .await
                        .map_err(|_| HtpbeError::Config("could not persist the resubmit key"))?;
                    key
                }
                other => return Ok((first.id, other)),
            }
        }
    };

    let second = client.check_pdf(url, name, &resubmit_key).await?;
    let route = match route_check(&second) {
        // Incomplete twice: unexamined, so never clean.
        Route::Resubmit => Route::RequestFromIssuer {
            reason: Some(StatusReason::AnalysisIncomplete),
        },
        other => other,
    };
    Ok((second.id, route))
}

async fn submit_document(
    State(state): State<AppState>,
    Json(req): Json<SubmitDocument>,
) -> (StatusCode, Json<CaseOutcome>) {
    let checked = run_check(&state.htpbe, &req).await;

    let (status, outcome) = match checked {
        Ok((check_id, Route::Proceed)) => (StatusCode::OK, CaseOutcome::Proceed { check_id }),
        Ok((check_id, Route::HumanReview { markers, confidence })) => (
            StatusCode::OK,
            CaseOutcome::HumanReview { check_id, markers, confidence },
        ),
        Ok((check_id, Route::RequestFromIssuer { reason })) => {
            (StatusCode::OK, CaseOutcome::RequestFromIssuer { check_id, reason })
        }
        // run_check never returns Resubmit; if it ever did, the file is unexamined.
        Ok((_, Route::Resubmit)) => (StatusCode::ACCEPTED, CaseOutcome::CheckPending),

        Err(err) => match err.code() {
            Some(code @ ("file_too_large" | "invalid_pdf")) => (
                StatusCode::UNPROCESSABLE_ENTITY,
                CaseOutcome::FileNotCheckable { error_code: code.to_owned() },
            ),
            _ => {
                // Transport failure, retries exhausted, 401/402/403, 503, or a caller
                // bug such as idempotency_key_fingerprint_mismatch. The document
                // was not checked: never Proceed, never adverse.
                if err.is_out_of_credits() {
                    eprintln!("htpbe: no credits left, alert billing");
                } else {
                    eprintln!("htpbe check failed: {err}");
                }
                (StatusCode::ACCEPTED, CaseOutcome::CheckPending)
            }
        },
    };

    (status, Json(outcome))
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let api_key = std::env::var("HTPBE_API_KEY")?;
    let state = AppState { htpbe: HtpbeClient::new(api_key)? };

    let app = Router::new()
        .route("/documents", post(submit_document))
        .with_state(state);

    let listener = tokio::net::TcpListener::bind("0.0.0.0:8080").await?;
    axum::serve(listener, app).await?;
    Ok(())
}

// src/upload.rs
use std::time::Duration;

use aws_sdk_s3::presigning::PresigningConfig;
use aws_sdk_s3::primitives::ByteStream;
use aws_sdk_s3::Client as S3Client;
use uuid::Uuid;

type BoxError = Box<dyn std::error::Error + Send + Sync>;

/// Covers one in-process check plus its retries, NOT the pending-case
/// window: a case retried after the URL expired is restaged instead.
const PRESIGN_TTL: Duration = Duration::from_secs(30 * 60);

/// Persist all three fields with the case. Every check of this case runs
/// from the stored URL and key while the URL is valid.
pub struct StagedCheck {
    pub object_key: String,
    pub document_url: String,
    pub check_key: String,
}

/// Uploads once, signs once, mints the Idempotency-Key once.
pub async fn stage_upload(s3: &S3Client, bucket: &str, pdf: Vec<u8>) -> Result<StagedCheck, BoxError> {
    let object_key = format!("incoming/{}.pdf", Uuid::new_v4());
    s3.put_object()
        .bucket(bucket)
        .key(&object_key)
        .content_type("application/pdf")
        .body(ByteStream::from(pdf))
        .send()
        .await?;
    sign(s3, bucket, object_key).await
}

/// Only for a check whose original attempt FAILED and whose URL has since
/// expired: a new URL is a new logical request, so it gets a new key too.
pub async fn restage(s3: &S3Client, bucket: &str, object_key: String) -> Result<StagedCheck, BoxError> {
    sign(s3, bucket, object_key).await
}

async fn sign(s3: &S3Client, bucket: &str, object_key: String) -> Result<StagedCheck, BoxError> {
    let presigned = s3
        .get_object()
        .bucket(bucket)
        .key(&object_key)
        .presigned(PresigningConfig::expires_in(PRESIGN_TTL)?)
        .await?;
    Ok(StagedCheck {
        object_key,
        document_url: presigned.uri().to_owned(),
        check_key: Uuid::new_v4().to_string(),
    })
}

#[derive(Clone)]
pub struct CheckJob {
    pub pdf_url: String,
    /// Minted at enqueue time and persisted, so a worker that picks the
    /// job up again replays the same logical request.
    pub idempotency_key: String,
}