24 changed files with 999 additions and 2547 deletions
+23 -17
View File
@@ -193,36 +193,42 @@ only as SHA-256 digests, and cannot mint/revoke tokens.
## OCSP responder ## OCSP responder
KCA can publish full per-CA certificate-status snapshots to any S3-compatible KCA publishes full per-CA certificate-status snapshots to Redis and ships a
bucket and ships a standalone Rust/Axum responder on port `2560`. standalone Rust/Axum responder on port `2560`. By default, snapshots use the
same Redis instance and logical database as sessions; setting
`KCA_OCSP_REDIS_URL` selects a dedicated OCSP instance instead.
- KCA publishes `ocsp/<KCA CA UUID>.json` whenever certificates are issued, - KCA publishes `ocsp:<KCA CA UUID>` whenever certificates are issued, renewed,
renewed, revoked, deleted or imported. Revocation reasons are revoked, deleted or imported. Revocation reasons are included; this is a full
included; this is a full status snapshot, not a revoked-only list. CBOR-encoded status snapshot, not a revoked-only list. OCSP keys are separate
- The responder has no Postgres/Redis/KCA master-key access. It only reads the from session keys through their prefix.
configured S3 prefix and uses a delegated OCSP signing certificate/key per - The responder has no Postgres or KCA master-key access. It reads only OCSP
CA. snapshots from Redis and uses a delegated OCSP signing certificate/key per
CA. Where Redis ACLs are available, restrict the responder to `ocsp:*` so it
cannot read session entries.
- Each CA snapshot is cached for one second with a single-flight refresh, so - Each CA snapshot is cached for one second with a single-flight refresh, so
concurrent requests cause at most one S3 `GetObject` per CA per second. concurrent requests share one atomic Redis `GET` + `PTTL` lookup per CA.
- The publisher refreshes unchanged snapshots every 30 seconds by default. The - The publisher refreshes unchanged snapshots every 30 seconds by default and
responder rejects snapshots older than its configured grace period even when sets each key's expiry to 60 seconds. A missing or expired snapshot produces
S3 is readable, and returns `tryLater` when no fresh snapshot is available. `unknown`; Redis failures produce `tryLater` when no unexpired cached
- OCSP `thisUpdate` reflects the snapshot generation time and `nextUpdate` never snapshot remains.
extends beyond the snapshot's hard expiry. - OCSP `thisUpdate` is the response time and `nextUpdate` never extends beyond
the snapshot's Redis expiry.
- Gitea Actions publishes `gitea.woggioni.net/woggioni/kca-ocsp` on the same - Gitea Actions publishes `gitea.woggioni.net/woggioni/kca-ocsp` on the same
`release/*` tags as the server image. The responder image is a statically `release/*` tags as the server image. The responder image is a statically
linked musl executable running `FROM scratch` with only the CA bundle. linked musl executable running `FROM scratch` with only the CA bundle.
Enable publishing in KCA with `KCA_OCSP_S3_BUCKET` and the other Configure `REDIS_URL`, optional `KCA_OCSP_REDIS_URL`, and the other
`KCA_OCSP_*` variables in `server/.env.example`. Run the responder with a `KCA_OCSP_*` variables in `server/.env.example`. Run the responder with a
configuration derived from `ocsp-responder/responder.example.yaml`, setting configuration derived from `ocsp-responder/responder.example.yaml`, setting
each issuer's `ca_id` to its CA certificate UUID in KCA, or with Compose: each issuer's `ca_id` to its CA certificate UUID in KCA and using the OCSP
Redis URL, or with Compose:
```sh ```sh
docker compose --profile ocsp up --build docker compose --profile ocsp up --build
``` ```
See `ocsp-responder/README.md` for the JSON layout, signer requirements and See `ocsp-responder/README.md` for the CBOR layout, signer requirements and
`openssl ocsp` verification. `openssl ocsp` verification.
## REST API ## REST API
+3 -1
View File
@@ -1,7 +1,9 @@
# KCA — Kubernetes deployment sketch (namespace: kca). # KCA — Kubernetes deployment sketch (namespace: kca).
# #
# Replace every REPLACE_ME. Postgres and the OIDC provider are external; Redis # Replace every REPLACE_ME. Postgres and the OIDC provider are external; Redis
# is included because sessions are disposable. KCA_MASTER_KEY is the AES key # stores disposable sessions and expiring OCSP snapshots. Set
# KCA_OCSP_REDIS_URL to publish OCSP snapshots to a separate instance instead.
# KCA_MASTER_KEY is the AES key
# protecting private keys at rest: generate it once and never lose it. # protecting private keys at rest: generate it once and never lose it.
# #
# kubectl apply -f deploy/k8s/kca.yaml # kubectl apply -f deploy/k8s/kca.yaml
+13 -17
View File
@@ -105,24 +105,20 @@ services:
OTEL_EXPORTER_OTLP_ENDPOINT: ${OTEL_EXPORTER_OTLP_ENDPOINT:-} OTEL_EXPORTER_OTLP_ENDPOINT: ${OTEL_EXPORTER_OTLP_ENDPOINT:-}
OTEL_EXPORTER_OTLP_HEADERS: ${OTEL_EXPORTER_OTLP_HEADERS:-} OTEL_EXPORTER_OTLP_HEADERS: ${OTEL_EXPORTER_OTLP_HEADERS:-}
OTEL_EXCLUDED_PATHS: ${OTEL_EXCLUDED_PATHS:-/api/health} OTEL_EXCLUDED_PATHS: ${OTEL_EXCLUDED_PATHS:-/api/health}
# OCSP status publisher (optional). The bucket must be writable by # OCSP snapshots default to the session Redis. Set KCA_OCSP_REDIS_URL to
# KCA; the responder later reads the same prefix with read-only access. # use a dedicated instance instead.
KCA_OCSP_S3_BUCKET: ${KCA_OCSP_S3_BUCKET:-} KCA_OCSP_REDIS_URL: ${KCA_OCSP_REDIS_URL:-}
KCA_OCSP_S3_PREFIX: ${KCA_OCSP_S3_PREFIX:-ocsp} KCA_OCSP_REDIS_PREFIX: ${KCA_OCSP_REDIS_PREFIX:-ocsp}
KCA_OCSP_S3_REGION: ${KCA_OCSP_S3_REGION:-}
KCA_OCSP_S3_ENDPOINT: ${KCA_OCSP_S3_ENDPOINT:-}
KCA_OCSP_S3_FORCE_PATH_STYLE: ${KCA_OCSP_S3_FORCE_PATH_STYLE:-false}
KCA_OCSP_PUBLISH_INTERVAL_SECONDS: ${KCA_OCSP_PUBLISH_INTERVAL_SECONDS:-1} KCA_OCSP_PUBLISH_INTERVAL_SECONDS: ${KCA_OCSP_PUBLISH_INTERVAL_SECONDS:-1}
KCA_OCSP_HEARTBEAT_INTERVAL_SECONDS: ${KCA_OCSP_HEARTBEAT_INTERVAL_SECONDS:-30} KCA_OCSP_HEARTBEAT_INTERVAL_SECONDS: ${KCA_OCSP_HEARTBEAT_INTERVAL_SECONDS:-30}
AWS_ACCESS_KEY_ID: ${AWS_ACCESS_KEY_ID:-} KCA_OCSP_REDIS_EXPIRY_SECONDS: ${KCA_OCSP_REDIS_EXPIRY_SECONDS:-60}
AWS_SECRET_ACCESS_KEY: ${AWS_SECRET_ACCESS_KEY:-}
AWS_SESSION_TOKEN: ${AWS_SESSION_TOKEN:-}
AWS_REGION: ${AWS_REGION:-}
ports: ports:
- "127.0.0.1:${APP_PORT:-8080}:8080" - "127.0.0.1:${APP_PORT:-8080}:8080"
# Optional standalone OCSP responder. It deliberately has no database, # Optional standalone OCSP responder. It has no database or KCA master-key
# Redis or KCA master-key environment. Enable with: # environment, but reads OCSP keys from the configured Redis (the session
# Redis by default).
# Enable with:
# docker compose --profile ocsp up --build # docker compose --profile ocsp up --build
ocsp: ocsp:
profiles: ["ocsp"] profiles: ["ocsp"]
@@ -130,12 +126,12 @@ services:
build: build:
context: . context: .
dockerfile: ocsp-responder/Dockerfile dockerfile: ocsp-responder/Dockerfile
depends_on:
redis:
condition: service_healthy
environment: environment:
OCSP_CONFIG: /etc/ocsp/responder.yaml OCSP_CONFIG: /etc/ocsp/responder.yaml
AWS_ACCESS_KEY_ID: ${AWS_ACCESS_KEY_ID:-} OCSP_REDIS_URL: ${OCSP_REDIS_URL:-redis://redis:6379/0}
AWS_SECRET_ACCESS_KEY: ${AWS_SECRET_ACCESS_KEY:-}
AWS_SESSION_TOKEN: ${AWS_SESSION_TOKEN:-}
AWS_REGION: ${AWS_REGION:-}
volumes: volumes:
- ${OCSP_CONFIG_FILE:-./ocsp-responder/responder.example.yaml}:/etc/ocsp/responder.yaml:ro - ${OCSP_CONFIG_FILE:-./ocsp-responder/responder.example.yaml}:/etc/ocsp/responder.yaml:ro
- ${OCSP_CA_DIR:-./ocsp-responder/ca}:/etc/ocsp/ca:ro - ${OCSP_CA_DIR:-./ocsp-responder/ca}:/etc/ocsp/ca:ro
+159 -1032
View File
File diff suppressed because it is too large Load Diff
+2 -12
View File
@@ -11,7 +11,7 @@ tower-http = { version = "0.6", features = ["limit", "trace", "timeout"] }
tracing = "0.1" tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] } tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] }
serde = { version = "1", features = ["derive"] } serde = { version = "1", features = ["derive"] }
serde_json = "1" cbor2 = "1"
serde_yaml = "0.9" serde_yaml = "0.9"
chrono = { version = "0.4", features = ["serde"] } chrono = { version = "0.4", features = ["serde"] }
thiserror = "2" thiserror = "2"
@@ -30,14 +30,4 @@ ecdsa = { version = "0.16", features = ["der"] }
pkcs8 = { version = "0.10", features = ["pem"] } pkcs8 = { version = "0.10", features = ["pem"] }
x509-ocsp = { version = "0.2.1", features = ["builder", "std"] } x509-ocsp = { version = "0.2.1", features = ["builder", "std"] }
x509-cert = { version = "0.2", features = ["std"] } x509-cert = { version = "0.2", features = ["std"] }
aws-config = { version = "1", default-features = false, features = [ redis = { version = "1", features = ["tokio-comp", "connection-manager", "script"] }
"behavior-version-latest",
"credentials-process",
"legacy-client",
"rt-tokio",
] }
aws-sdk-s3 = { version = "1", default-features = false, features = [
"http-1x",
"legacy-https-client",
"rt-tokio",
] }
+3 -3
View File
@@ -2,9 +2,9 @@
# Standalone Rust/Axum OCSP responder. # Standalone Rust/Axum OCSP responder.
# #
# The final image is scratch: the binary is a statically linked musl # The final image is scratch: the binary is a statically linked musl
# executable and the only file besides it is the CA bundle needed for # executable and the only file besides it is the CA bundle needed for TLS
# outbound HTTPS to S3. Configuration, CA certificates and delegated signer # connections. Configuration, CA certificates and delegated signer material
# material are mounted at runtime. # are mounted at runtime.
FROM rust:1-alpine AS builder FROM rust:1-alpine AS builder
+48 -45
View File
@@ -1,50 +1,51 @@
# KCA OCSP responder # KCA OCSP responder
Standalone Rust/Axum responder. It reads one full JSON status snapshot per CA Standalone Rust/Axum responder. It reads one full CBOR status snapshot per CA
from S3-compatible object storage, caches it for one second with a single-flight from Redis, caches it for one second with a single-flight refresh, and signs
refresh, and signs OCSP responses with a delegated responder certificate. OCSP responses with a delegated responder certificate.
It has **no KCA database access**, no Redis access, and no KCA master key. It It has **no KCA database access** and no KCA master key. It only needs:
only needs:
- `s3:GetObject` on the configured bucket/prefix; - read access to OCSP keys in the Redis instance selected by KCA (the session
Redis by default, or a dedicated instance when `KCA_OCSP_REDIS_URL` is set);
- the CA certificate and its KCA CA UUID for each issuer; - the CA certificate and its KCA CA UUID for each issuer;
- a delegated OCSP signer certificate and private key for each CA. The - a delegated OCSP signer certificate and private key for each CA. The
responder currently accepts RSA-SHA256 and ECDSA P-256 signers in PKCS#8, responder currently accepts RSA-SHA256 and ECDSA P-256 signers in PKCS#8,
PKCS#1 RSA, or SEC1 EC PEM form. PKCS#1 RSA, or SEC1 EC PEM form.
Where Redis ACLs are available, restrict the responder's credentials to the
OCSP prefix so it cannot read session entries. When sharing an instance, KCA
needs access to both `ocsp:*` and `kaya:session:*`; a dedicated OCSP instance
can use separate credentials.
## Snapshot layout ## Snapshot layout
KCA publishes objects as: KCA publishes keys as:
``` ```
<prefix>/<KCA CA UUID>.json <prefix>:<KCA CA UUID>
``` ```
Set each issuer's required `ca_id` to the UUID of its CA certificate row in KCA. The default prefix is `ocsp`. Set each issuer's required `ca_id` to the UUID of
This isolates snapshots when the same public CA certificate exists in multiple its CA certificate row in KCA. This isolates snapshots when the same public CA
vaults. The responder validates both this ID and the CA certificate fingerprint certificate exists in multiple vaults. The responder validates both this ID and
against every downloaded snapshot. the CA certificate fingerprint against every downloaded snapshot.
KCA publishes these objects when `KCA_OCSP_S3_BUCKET` is configured. The JSON KCA publishes these keys when either `REDIS_URL` or `KCA_OCSP_REDIS_URL` is
contains all certificates issued by that CA, both valid and revoked, and a configured, preferring the dedicated OCSP URL when both are present. The
required `generated_at` timestamp. The responder never infers status from a `kca-ocsp-status/v2` payload is encoded with CBOR and contains all certificates
partial revoked-only list. issued by that CA, both valid and revoked, without a generation timestamp.
Freshness is carried by the Redis key TTL. The responder never infers status
## Upgrading from fingerprint-named snapshots from a partial revoked-only list. Publisher and responder upgrades must be
coordinated because legacy JSON payloads are rejected until republished.
Deploy the updated responder with `ca_id` before or together with the updated
KCA publisher. The responder deliberately does not fall back to legacy
`<prefix>/<CA certificate SHA-256>.json` objects. Until KCA publishes the new
objects, OCSP requests return `tryLater`. After verifying the new snapshots,
the obsolete fingerprint-named objects can be removed.
## Run locally ## Run locally
1. Create a config based on `responder.example.yaml`, setting each issuer's 1. Create a config based on `responder.example.yaml`, setting each issuer's
`ca_id` to its CA certificate UUID in KCA. `ca_id` to its CA certificate UUID in KCA.
2. Put your CA certificates and delegated signer files where the config points. 2. Set `redis.url` or `OCSP_REDIS_URL` to the OCSP Redis URL used by KCA (the
3. Export standard AWS credentials / endpoint variables as needed. session `REDIS_URL` unless `KCA_OCSP_REDIS_URL` overrides it).
3. Put your CA certificates and delegated signer files where the config points.
4. Make the signer key readable by the container user (`65532:65532` in the 4. Make the signer key readable by the container user (`65532:65532` in the
published scratch image), or mount it via a Docker secret. published scratch image), or mount it via a Docker secret.
5. Run: 5. Run:
@@ -62,12 +63,12 @@ docker run --rm -p 2560:2560 \
-v "$PWD/responder.yaml:/etc/ocsp/responder.yaml:ro" \ -v "$PWD/responder.yaml:/etc/ocsp/responder.yaml:ro" \
-v "$PWD/ca:/etc/ocsp/ca:ro" \ -v "$PWD/ca:/etc/ocsp/ca:ro" \
-v "$PWD/signers:/etc/ocsp/signers:ro" \ -v "$PWD/signers:/etc/ocsp/signers:ro" \
-e AWS_ACCESS_KEY_ID -e AWS_SECRET_ACCESS_KEY \ -e OCSP_REDIS_URL \
gitea.woggioni.net/woggioni/kca-ocsp:latest gitea.woggioni.net/woggioni/kca-ocsp:latest
``` ```
It runs `FROM scratch` and contains only a statically linked musl executable It runs `FROM scratch` and contains only a statically linked musl executable
and a CA bundle for outbound S3 HTTPS. There is no shell or package manager in and a CA bundle for TLS connections. There is no shell or package manager in
the image. the image.
## Verify with OpenSSL ## Verify with OpenSSL
@@ -96,21 +97,23 @@ build a trust store containing the CA or validate with your normal PKI tooling.
## Cache and availability ## Cache and availability
- Per-CA cache TTL: `s3.cache_ttl_ms` (default 1000 ms). - Per-CA cache TTL: `redis.cache_ttl_ms` (default 1000 ms).
- Concurrent requests for the same CA share one S3 `GetObject`. - Concurrent requests for the same CA share one refresh. A Redis Lua script
- `s3.stale_grace_seconds` is also the hard maximum snapshot age, measured from performs `GET` and `PTTL` atomically, keeping the snapshot and its expiry
`generated_at`. Expired or future-dated snapshots are rejected even when S3 consistent.
remains readable; if no valid cached snapshot remains, OCSP returns - A missing or expired key produces `unknown`, as does a serial absent from an
`tryLater`. existing snapshot. A key that exists without a TTL is rejected as a data
- If a refresh fails, a cached snapshot is served only while both its time since error. Redis connection and command failures produce `tryLater` when no
the last successful fetch and its `generated_at` age remain within unexpired cached snapshot remains.
`s3.stale_grace_seconds`. - If a refresh fails, a cached snapshot is served only until its Redis expiry.
A missing key clears cached status so it cannot be revived by a later Redis
failure.
- KCA republishes unchanged snapshots every - KCA republishes unchanged snapshots every
`KCA_OCSP_HEARTBEAT_INTERVAL_SECONDS` (default 30 seconds). Keep that interval `KCA_OCSP_HEARTBEAT_INTERVAL_SECONDS` (default 30 seconds) and sets each
comfortably below the responder's `s3.stale_grace_seconds` (default 60). For key's expiry using `KCA_OCSP_REDIS_EXPIRY_SECONDS` (default 60 seconds).
a rolling upgrade, enable publisher heartbeats before the freshness-enforcing Keep the heartbeat comfortably below the key expiry.
responder so existing snapshots remain within the age limit. - OCSP `thisUpdate` and `producedAt` use the response time. `nextUpdate` is
- OCSP `thisUpdate` is the snapshot's `generated_at`. `nextUpdate` is capped at capped at the snapshot's Redis expiry when a snapshot is present.
the snapshot's hard expiry, while `producedAt` is the response signing time. - With multiple CAs, the cache TTL is per CA. If you need a global
- With multiple CAs, the TTL is per CA. If you need a global one-request-per- one-request-per-second limit across all CAs, that requires a shared rate
second limit across all CAs, that requires a shared rate limiter. limiter.
+4 -8
View File
@@ -3,16 +3,12 @@
listen: 0.0.0.0:2560 listen: 0.0.0.0:2560
next_update_seconds: 10 next_update_seconds: 10
s3: redis:
bucket: kca-ocsp # OCSP Redis used by KCA (the session Redis unless KCA_OCSP_REDIS_URL is set).
# OCSP_REDIS_URL overrides this value at runtime.
url: redis://redis:6379/0
prefix: ocsp prefix: ocsp
region: eu-central-1
# Optional for MinIO / other S3-compatible stores:
# endpoint: https://s3.example.com
# force_path_style: true
cache_ttl_ms: 1000 cache_ttl_ms: 1000
# Hard maximum snapshot age. Keep comfortably above KCA's heartbeat interval.
stale_grace_seconds: 60
issuers: issuers:
- name: Example Root CA - name: Example Root CA
+112 -121
View File
@@ -1,9 +1,8 @@
//! One-second, single-flight cache of per-CA status snapshots. //! One-second, single-flight cache of per-CA status snapshots.
use std::sync::Arc; use std::sync::Arc;
use std::time::{Duration, Instant}; use std::time::{Duration, Instant, SystemTime};
use chrono::Utc;
use tokio::sync::{Mutex, RwLock}; use tokio::sync::{Mutex, RwLock};
use crate::snapshot::Snapshot; use crate::snapshot::Snapshot;
@@ -11,7 +10,8 @@ use crate::store::ObjectStore;
#[derive(Clone)] #[derive(Clone)]
pub struct CachedSnapshot { pub struct CachedSnapshot {
snapshot: Arc<Snapshot>, pub snapshot: Arc<Snapshot>,
pub expires_at: SystemTime,
fetched_at: Instant, fetched_at: Instant,
} }
@@ -26,10 +26,10 @@ impl SnapshotCache {
Self::default() Self::default()
} }
/// Return a snapshot, refreshing from S3 at most once per `ttl`. /// Return a snapshot, refreshing from Redis at most once per `ttl`.
/// ///
/// Snapshots older than `stale_grace` are never used. On refresh failure, /// A cached snapshot is never used after its Redis key expires,
/// cached data is also limited to `stale_grace` since its last fetch. /// including when a refresh fails.
pub async fn get<S: ObjectStore + ?Sized>( pub async fn get<S: ObjectStore + ?Sized>(
&self, &self,
store: &S, store: &S,
@@ -37,12 +37,10 @@ impl SnapshotCache {
expected_sha256: &str, expected_sha256: &str,
expected_ca_id: &str, expected_ca_id: &str,
ttl: Duration, ttl: Duration,
stale_grace: Duration, ) -> Result<CachedSnapshot, CacheError> {
) -> Result<Arc<Snapshot>, CacheError> {
if let Some(value) = self.cached.read().await.as_ref() { if let Some(value) = self.cached.read().await.as_ref() {
if value.fetched_at.elapsed() < ttl && snapshot_age_valid(&value.snapshot, stale_grace) if fresh_cache_entry_valid(value, ttl) {
{ return Ok(value.clone());
return Ok(value.snapshot.clone());
} }
} }
@@ -52,10 +50,8 @@ impl SnapshotCache {
let stale = { let stale = {
let cache = self.cached.read().await; let cache = self.cached.read().await;
if let Some(value) = cache.as_ref() { if let Some(value) = cache.as_ref() {
if value.fetched_at.elapsed() < ttl if fresh_cache_entry_valid(value, ttl) {
&& snapshot_age_valid(&value.snapshot, stale_grace) return Ok(value.clone());
{
return Ok(value.snapshot.clone());
} }
Some(value.clone()) Some(value.clone())
} else { } else {
@@ -64,21 +60,31 @@ impl SnapshotCache {
}; };
match store.get(key).await { match store.get(key).await {
Ok(bytes) => match Snapshot::parse(&bytes, expected_sha256, expected_ca_id) { Ok(crate::store::StoredSnapshot::Missing) => {
Ok(snapshot) => match validate_snapshot_age(&snapshot, stale_grace) { // A key that expired or was deleted must not be replaced by
Ok(()) => { // an older cached status on a later Redis failure.
let snapshot = Arc::new(snapshot); *self.cached.write().await = None;
*self.cached.write().await = Some(CachedSnapshot { Err(CacheError::Missing)
snapshot: snapshot.clone(), }
Ok(crate::store::StoredSnapshot::Present { body, expires_at }) => {
let parsed = validate_expiry(expires_at).and_then(|()| {
Snapshot::parse(&body, expected_sha256, expected_ca_id)
.map_err(|error| error.to_string())
});
match parsed {
Ok(snapshot) => {
let cached = CachedSnapshot {
snapshot: Arc::new(snapshot),
expires_at,
fetched_at: Instant::now(), fetched_at: Instant::now(),
}); };
Ok(snapshot) *self.cached.write().await = Some(cached.clone());
Ok(cached)
} }
Err(error) => self.fallback(stale, error, stale_grace), Err(error) => self.fallback(stale, error),
}, }
Err(error) => self.fallback(stale, error.to_string(), stale_grace), }
}, Err(error) => self.fallback(stale, error.to_string()),
Err(error) => self.fallback(stale, error.to_string(), stale_grace),
} }
} }
@@ -86,45 +92,37 @@ impl SnapshotCache {
&self, &self,
stale: Option<CachedSnapshot>, stale: Option<CachedSnapshot>,
error: String, error: String,
stale_grace: Duration, ) -> Result<CachedSnapshot, CacheError> {
) -> Result<Arc<Snapshot>, CacheError> {
if let Some(value) = stale { if let Some(value) = stale {
if value.fetched_at.elapsed() <= stale_grace if expiry_valid(value.expires_at) {
&& snapshot_age_valid(&value.snapshot, stale_grace) tracing::warn!(error = %error, "serving unexpired cached OCSP snapshot after refresh failure");
{ return Ok(value);
tracing::warn!(error = %error, "serving stale OCSP snapshot");
return Ok(value.snapshot);
} }
} }
Err(CacheError::Unavailable(error)) Err(CacheError::Unavailable(error))
} }
} }
fn snapshot_age_valid(snapshot: &Snapshot, stale_grace: Duration) -> bool { fn fresh_cache_entry_valid(value: &CachedSnapshot, ttl: Duration) -> bool {
validate_snapshot_age(snapshot, stale_grace).is_ok() value.fetched_at.elapsed() < ttl && expiry_valid(value.expires_at)
} }
fn validate_snapshot_age(snapshot: &Snapshot, stale_grace: Duration) -> Result<(), String> { fn expiry_valid(expires_at: SystemTime) -> bool {
let age = Utc::now() expires_at > SystemTime::now()
.signed_duration_since(snapshot.generated_at) }
.to_std()
.map_err(|_| { fn validate_expiry(expires_at: SystemTime) -> Result<(), String> {
format!( if expiry_valid(expires_at) {
"snapshot generation time {} is in the future", Ok(())
snapshot.generated_at } else {
) Err(format!("snapshot Redis key expired at {expires_at:?}"))
})?;
if age > stale_grace {
return Err(format!(
"snapshot generated at {} exceeds maximum age {stale_grace:?}",
snapshot.generated_at
));
} }
Ok(())
} }
#[derive(Debug, thiserror::Error)] #[derive(Debug, thiserror::Error)]
pub enum CacheError { pub enum CacheError {
#[error("OCSP snapshot is missing")]
Missing,
#[error("OCSP snapshot unavailable: {0}")] #[error("OCSP snapshot unavailable: {0}")]
Unavailable(String), Unavailable(String),
} }
@@ -137,119 +135,97 @@ mod tests {
use bytes::Bytes; use bytes::Bytes;
use super::*; use super::*;
use crate::snapshot::sha256_hex; use crate::snapshot::{sha256_hex, test_body as body};
use crate::store::{ObjectStore, StoreError}; use crate::store::{ObjectStore, StoreError, StoredSnapshot};
struct FakeStore { struct FakeStore {
body: String, body: Vec<u8>,
expires_at: SystemTime,
calls: AtomicUsize, calls: AtomicUsize,
fail: AtomicBool, fail: AtomicBool,
missing: AtomicBool,
} }
impl FakeStore { impl FakeStore {
fn new(body: String) -> Self { fn new(body: Vec<u8>, lifetime: Duration) -> Self {
Self { Self {
body, body,
expires_at: SystemTime::now() + lifetime,
calls: AtomicUsize::new(0), calls: AtomicUsize::new(0),
fail: AtomicBool::new(false), fail: AtomicBool::new(false),
missing: AtomicBool::new(false),
}
}
fn expired(body: Vec<u8>) -> Self {
Self {
body,
expires_at: SystemTime::now() - Duration::from_secs(1),
calls: AtomicUsize::new(0),
fail: AtomicBool::new(false),
missing: AtomicBool::new(false),
} }
} }
} }
#[async_trait] #[async_trait]
impl ObjectStore for FakeStore { impl ObjectStore for FakeStore {
async fn get(&self, _key: &str) -> Result<Bytes, StoreError> { async fn get(&self, _key: &str) -> Result<StoredSnapshot, StoreError> {
self.calls.fetch_add(1, Ordering::SeqCst); self.calls.fetch_add(1, Ordering::SeqCst);
if self.fail.load(Ordering::SeqCst) { if self.fail.load(Ordering::SeqCst) {
return Err(StoreError::S3("offline".to_string())); return Err(StoreError::Redis("offline".to_string()));
} }
Ok(Bytes::from(self.body.clone())) if self.missing.load(Ordering::SeqCst) {
return Ok(StoredSnapshot::Missing);
}
Ok(StoredSnapshot::Present {
body: Bytes::from(self.body.clone()),
expires_at: self.expires_at,
})
} }
} }
fn body(sha: &str, generated_at: chrono::DateTime<Utc>) -> String {
let generated_at = generated_at.to_rfc3339();
format!(
r#"{{"schema":"kca-ocsp-status/v1","issuer":{{"id":"ca","sha256":"{sha}"}},"generated_at":"{generated_at}","entries":[{{"serial_hex":"01","status":"valid"}}]}}"#
)
}
#[tokio::test] #[tokio::test]
async fn single_flight_refreshes_once() { async fn single_flight_refreshes_once() {
let sha = sha256_hex(b"ca"); let sha = sha256_hex(b"ca");
let store = FakeStore::new(body(&sha, Utc::now())); let store = FakeStore::new(body(&sha), Duration::from_secs(1));
let cache = SnapshotCache::new(); let cache = SnapshotCache::new();
let (a, b) = tokio::join!( let (a, b) = tokio::join!(
cache.get( cache.get(&store, "key", &sha, "ca", Duration::from_millis(100),),
&store, cache.get(&store, "key", &sha, "ca", Duration::from_millis(100),)
"key",
&sha,
"ca",
Duration::from_millis(100),
Duration::from_secs(1)
),
cache.get(
&store,
"key",
&sha,
"ca",
Duration::from_millis(100),
Duration::from_secs(1)
)
); );
assert!(a.is_ok()); assert!(a.is_ok());
assert!(b.is_ok()); assert!(b.is_ok());
assert_eq!(store.calls.load(Ordering::SeqCst), 1); assert_eq!(store.calls.load(Ordering::SeqCst), 1);
assert_eq!(a.unwrap().expires_at, store.expires_at);
} }
#[tokio::test] #[tokio::test]
async fn refreshes_after_ttl_and_serves_stale_on_error() { async fn refreshes_after_ttl_and_serves_unexpired_cache_on_error() {
let sha = sha256_hex(b"ca"); let sha = sha256_hex(b"ca");
let store = FakeStore::new(body(&sha, Utc::now())); let store = FakeStore::new(body(&sha), Duration::from_secs(1));
let cache = SnapshotCache::new(); let cache = SnapshotCache::new();
cache cache
.get( .get(&store, "key", &sha, "ca", Duration::from_millis(20))
&store,
"key",
&sha,
"ca",
Duration::from_millis(20),
Duration::from_secs(1),
)
.await .await
.unwrap(); .unwrap();
tokio::time::sleep(Duration::from_millis(30)).await; tokio::time::sleep(Duration::from_millis(30)).await;
store.fail.store(true, Ordering::SeqCst); store.fail.store(true, Ordering::SeqCst);
let stale = cache let stale = cache
.get( .get(&store, "key", &sha, "ca", Duration::from_millis(20))
&store,
"key",
&sha,
"ca",
Duration::from_millis(20),
Duration::from_secs(1),
)
.await .await
.unwrap(); .unwrap();
assert_eq!(stale.entries.len(), 1); assert_eq!(stale.snapshot.entries.len(), 1);
assert_eq!(store.calls.load(Ordering::SeqCst), 2); assert_eq!(store.calls.load(Ordering::SeqCst), 2);
} }
#[tokio::test] #[tokio::test]
async fn rejects_expired_snapshot_from_healthy_store() { async fn rejects_expired_snapshot_from_healthy_store() {
let sha = sha256_hex(b"ca"); let sha = sha256_hex(b"ca");
let generated_at = Utc::now() - chrono::Duration::seconds(30); let store = FakeStore::expired(body(&sha));
let store = FakeStore::new(body(&sha, generated_at));
let cache = SnapshotCache::new(); let cache = SnapshotCache::new();
let result = cache let result = cache
.get( .get(&store, "key", &sha, "ca", Duration::from_millis(1))
&store,
"key",
&sha,
"ca",
Duration::from_millis(1),
Duration::from_secs(1),
)
.await; .await;
assert!(matches!(result, Err(CacheError::Unavailable(_)))); assert!(matches!(result, Err(CacheError::Unavailable(_))));
assert_eq!(store.calls.load(Ordering::SeqCst), 1); assert_eq!(store.calls.load(Ordering::SeqCst), 1);
@@ -258,22 +234,37 @@ mod tests {
#[tokio::test] #[tokio::test]
async fn successful_fetches_do_not_extend_snapshot_lifetime() { async fn successful_fetches_do_not_extend_snapshot_lifetime() {
let sha = sha256_hex(b"ca"); let sha = sha256_hex(b"ca");
let store = FakeStore::new(body(&sha, Utc::now())); let store = FakeStore::new(body(&sha), Duration::from_millis(100));
let cache = SnapshotCache::new(); let cache = SnapshotCache::new();
let ttl = Duration::from_secs(1); let ttl = Duration::from_secs(1);
let stale_grace = Duration::from_millis(100); cache.get(&store, "key", &sha, "ca", ttl).await.unwrap();
cache
.get(&store, "key", &sha, "ca", ttl, stale_grace)
.await
.unwrap();
tokio::time::sleep(Duration::from_millis(120)).await; tokio::time::sleep(Duration::from_millis(120)).await;
let expired = cache.get(&store, "key", &sha, "ca", ttl, stale_grace).await; let expired = cache.get(&store, "key", &sha, "ca", ttl).await;
assert!(matches!(expired, Err(CacheError::Unavailable(_)))); assert!(matches!(expired, Err(CacheError::Unavailable(_))));
store.fail.store(true, Ordering::SeqCst); store.fail.store(true, Ordering::SeqCst);
let unavailable = cache.get(&store, "key", &sha, "ca", ttl, stale_grace).await; let unavailable = cache.get(&store, "key", &sha, "ca", ttl).await;
assert!(matches!(unavailable, Err(CacheError::Unavailable(_)))); assert!(matches!(unavailable, Err(CacheError::Unavailable(_))));
assert_eq!(store.calls.load(Ordering::SeqCst), 3); assert_eq!(store.calls.load(Ordering::SeqCst), 3);
} }
#[tokio::test]
async fn missing_key_clears_cached_status_instead_of_reviving_it() {
let sha = sha256_hex(b"ca");
let store = FakeStore::new(body(&sha), Duration::from_secs(1));
let cache = SnapshotCache::new();
let ttl = Duration::from_millis(20);
cache.get(&store, "key", &sha, "ca", ttl).await.unwrap();
store.missing.store(true, Ordering::SeqCst);
tokio::time::sleep(Duration::from_millis(30)).await;
let missing = cache.get(&store, "key", &sha, "ca", ttl).await;
assert!(matches!(missing, Err(CacheError::Missing)));
store.missing.store(false, Ordering::SeqCst);
store.fail.store(true, Ordering::SeqCst);
let unavailable = cache.get(&store, "key", &sha, "ca", ttl).await;
assert!(matches!(unavailable, Err(CacheError::Unavailable(_))));
}
} }
+19 -30
View File
@@ -10,7 +10,7 @@ pub struct Config {
#[serde(default = "default_listen")] #[serde(default = "default_listen")]
pub listen: SocketAddr, pub listen: SocketAddr,
#[serde(default)] #[serde(default)]
pub s3: S3Config, pub redis: RedisConfig,
#[serde(default)] #[serde(default)]
pub issuers: Vec<IssuerConfig>, pub issuers: Vec<IssuerConfig>,
#[serde(default = "default_ocsp_next_update_seconds")] #[serde(default = "default_ocsp_next_update_seconds")]
@@ -27,42 +27,32 @@ pub struct IssuerConfig {
} }
#[derive(Debug, Clone, Deserialize)] #[derive(Debug, Clone, Deserialize)]
pub struct S3Config { pub struct RedisConfig {
pub bucket: String, pub url: String,
#[serde(default = "default_prefix")] #[serde(default = "default_prefix")]
pub prefix: String, pub prefix: String,
pub region: Option<String>,
pub endpoint: Option<String>,
#[serde(default)]
pub force_path_style: bool,
#[serde(default = "default_cache_ttl_ms")] #[serde(default = "default_cache_ttl_ms")]
pub cache_ttl_ms: u64, pub cache_ttl_ms: u64,
#[serde(default = "default_stale_grace_seconds")]
pub stale_grace_seconds: u64,
} }
impl Default for S3Config { impl Default for RedisConfig {
fn default() -> Self { fn default() -> Self {
Self { Self {
bucket: String::new(), url: String::new(),
prefix: default_prefix(), prefix: default_prefix(),
region: None,
endpoint: None,
force_path_style: false,
cache_ttl_ms: default_cache_ttl_ms(), cache_ttl_ms: default_cache_ttl_ms(),
stale_grace_seconds: default_stale_grace_seconds(),
} }
} }
} }
impl S3Config { impl RedisConfig {
pub fn object_key(&self, ca_id: &str) -> String { pub fn key(&self, ca_id: &str) -> String {
let prefix = self.prefix.trim_matches('/'); let prefix = self.prefix.trim_matches(['/', ':']);
let name = format!("{}.json", ca_id.to_lowercase()); let name = ca_id.to_lowercase();
if prefix.is_empty() { if prefix.is_empty() {
name name
} else { } else {
format!("{prefix}/{name}") format!("{prefix}:{name}")
} }
} }
} }
@@ -81,10 +71,6 @@ fn default_cache_ttl_ms() -> u64 {
1000 1000
} }
fn default_stale_grace_seconds() -> u64 {
60
}
fn default_ocsp_next_update_seconds() -> u64 { fn default_ocsp_next_update_seconds() -> u64 {
10 10
} }
@@ -97,8 +83,11 @@ pub fn load_config() -> Result<Config, Box<dyn std::error::Error + Send + Sync>>
if config.issuers.is_empty() { if config.issuers.is_empty() {
return Err("configuration must contain at least one issuer".into()); return Err("configuration must contain at least one issuer".into());
} }
if config.s3.bucket.trim().is_empty() { if let Ok(url) = std::env::var("OCSP_REDIS_URL") {
return Err("s3.bucket must be configured".into()); config.redis.url = url;
}
if config.redis.url.trim().is_empty() {
return Err("redis.url must be configured".into());
} }
if let Ok(listen) = std::env::var("OCSP_LISTEN") { if let Ok(listen) = std::env::var("OCSP_LISTEN") {
config.listen = listen.parse()?; config.listen = listen.parse()?;
@@ -111,14 +100,14 @@ mod tests {
use super::*; use super::*;
#[test] #[test]
fn object_key_uses_ca_id() { fn redis_key_uses_ca_id() {
let config = S3Config { let config = RedisConfig {
prefix: "ocsp".to_string(), prefix: "ocsp".to_string(),
..Default::default() ..Default::default()
}; };
assert_eq!( assert_eq!(
config.object_key("123E4567-E89B-12D3-A456-426614174000"), config.key("123E4567-E89B-12D3-A456-426614174000"),
"ocsp/123e4567-e89b-12d3-a456-426614174000.json" "ocsp:123e4567-e89b-12d3-a456-426614174000"
); );
} }
} }
+3 -4
View File
@@ -111,9 +111,9 @@ impl Issuer {
pub async fn snapshot<S: crate::store::ObjectStore + ?Sized>( pub async fn snapshot<S: crate::store::ObjectStore + ?Sized>(
&self, &self,
store: &S, store: &S,
config: &crate::config::S3Config, config: &crate::config::RedisConfig,
) -> Result<Arc<crate::snapshot::Snapshot>, crate::cache::CacheError> { ) -> Result<crate::cache::CachedSnapshot, crate::cache::CacheError> {
let key = config.object_key(&self.ca_id); let key = config.key(&self.ca_id);
self.cache self.cache
.get( .get(
store, store,
@@ -121,7 +121,6 @@ impl Issuer {
&self.fingerprint, &self.fingerprint,
&self.ca_id, &self.ca_id,
Duration::from_millis(config.cache_ttl_ms), Duration::from_millis(config.cache_ttl_ms),
Duration::from_secs(config.stale_grace_seconds),
) )
.await .await
} }
+1 -1
View File
@@ -32,7 +32,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
.init(); .init();
let config = config::load_config()?; let config = config::load_config()?;
let store = StatusStore::new(&config.s3).await; let store = StatusStore::new(&config.redis).await?;
let mut issuers = Vec::with_capacity(config.issuers.len()); let mut issuers = Vec::with_capacity(config.issuers.len());
for issuer_config in &config.issuers { for issuer_config in &config.issuers {
let issuer = Issuer::load(issuer_config)?; let issuer = Issuer::load(issuer_config)?;
+25 -29
View File
@@ -8,7 +8,6 @@ use axum::extract::{Path, State};
use axum::http::{header, HeaderValue, StatusCode}; use axum::http::{header, HeaderValue, StatusCode};
use axum::response::{IntoResponse, Response}; use axum::response::{IntoResponse, Response};
use base64::Engine; use base64::Engine;
use chrono::{DateTime, Utc};
use der::{Decode, Encode}; use der::{Decode, Encode};
use x509_cert::ext::pkix::CrlReason; use x509_cert::ext::pkix::CrlReason;
use x509_ocsp::builder::OcspResponseBuilder; use x509_ocsp::builder::OcspResponseBuilder;
@@ -67,8 +66,11 @@ async fn handle_request(state: &AppState, body: &[u8]) -> Response {
return ocsp_response(OcspResponse::unauthorized()); return ocsp_response(OcspResponse::unauthorized());
} }
let snapshot = match issuer.snapshot(&state.store, &state.config.s3).await { let (snapshot, expires_at) = match issuer.snapshot(&state.store, &state.config.redis).await {
Ok(snapshot) => snapshot, Ok(cached) => (cached.snapshot, Some(cached.expires_at)),
Err(crate::cache::CacheError::Missing) => {
(Arc::new(crate::snapshot::Snapshot::default()), None)
}
Err(error) => { Err(error) => {
tracing::warn!(issuer = %issuer.name, error = %error, "OCSP snapshot unavailable"); tracing::warn!(issuer = %issuer.name, error = %error, "OCSP snapshot unavailable");
return ocsp_response(OcspResponse::try_later()); return ocsp_response(OcspResponse::try_later());
@@ -83,12 +85,8 @@ async fn handle_request(state: &AppState, body: &[u8]) -> Response {
return ocsp_response(OcspResponse::internal_error()); return ocsp_response(OcspResponse::internal_error());
} }
}; };
let (this_update, next_update) = response_update_times( let (this_update, next_update) =
snapshot.generated_at, response_update_times(expires_at, state.config.next_update_seconds, now);
state.config.next_update_seconds,
state.config.s3.stale_grace_seconds,
now,
);
let this_update = match OcspGeneralizedTime::try_from(this_update) { let this_update = match OcspGeneralizedTime::try_from(this_update) {
Ok(value) => value, Ok(value) => value,
Err(error) => { Err(error) => {
@@ -143,17 +141,17 @@ async fn handle_request(state: &AppState, body: &[u8]) -> Response {
} }
fn response_update_times( fn response_update_times(
generated_at: DateTime<Utc>, expires_at: Option<SystemTime>,
next_update_seconds: u64, next_update_seconds: u64,
stale_grace_seconds: u64,
now: SystemTime, now: SystemTime,
) -> (SystemTime, Option<SystemTime>) { ) -> (SystemTime, Option<SystemTime>) {
let this_update = system_time(generated_at);
let expires_at = this_update.checked_add(Duration::from_secs(stale_grace_seconds));
let next_update = now let next_update = now
.checked_add(Duration::from_secs(next_update_seconds)) .checked_add(Duration::from_secs(next_update_seconds))
.and_then(|next| expires_at.map(|expires| next.min(expires))); .map(|next| match expires_at {
(this_update, next_update) Some(expires_at) => next.min(expires_at),
None => next,
});
(now, next_update)
} }
fn ocsp_response(response: OcspResponse) -> Response { fn ocsp_response(response: OcspResponse) -> Response {
@@ -179,22 +177,20 @@ mod tests {
use super::*; use super::*;
#[test] #[test]
fn update_times_use_snapshot_age_and_bound_next_update() { fn update_times_use_response_time_and_bound_next_update_to_expiry() {
let generated_at = DateTime::parse_from_rfc3339("2026-01-01T00:00:00Z") let base = SystemTime::UNIX_EPOCH + Duration::from_secs(1_767_225_600);
.unwrap() let expires_at = base + Duration::from_secs(60);
.with_timezone(&Utc);
let generated_at_system = system_time(generated_at);
let now = generated_at_system + Duration::from_secs(5); let now = base + Duration::from_secs(5);
let (this_update, next_update) = response_update_times(generated_at, 10, 60, now); let (this_update, next_update) = response_update_times(Some(expires_at), 10, now);
assert_eq!(this_update, generated_at_system); assert_eq!(this_update, now);
assert_eq!(next_update, Some(now + Duration::from_secs(10))); assert_eq!(next_update, Some(now + Duration::from_secs(10)));
let now = generated_at_system + Duration::from_secs(55); let now = base + Duration::from_secs(55);
let (_, next_update) = response_update_times(generated_at, 10, 60, now); let (_, next_update) = response_update_times(Some(expires_at), 10, now);
assert_eq!( assert_eq!(next_update, Some(expires_at));
next_update,
Some(generated_at_system + Duration::from_secs(60)) let (_, next_update) = response_update_times(None, 10, now);
); assert_eq!(next_update, Some(now + Duration::from_secs(10)));
} }
} }
+74 -51
View File
@@ -3,12 +3,12 @@
use std::collections::HashMap; use std::collections::HashMap;
use chrono::{DateTime, Utc}; use chrono::{DateTime, Utc};
use serde::Deserialize; use serde::{Deserialize, Serialize};
use sha1::{Digest as _, Sha1}; use sha1::{Digest as _, Sha1};
use sha2::Sha256; use sha2::Sha256;
use x509_cert::ext::pkix::CrlReason; use x509_cert::ext::pkix::CrlReason;
pub const SCHEMA: &str = "kca-ocsp-status/v1"; pub const SCHEMA: &str = "kca-ocsp-status/v2";
#[derive(Debug, Clone, Copy, PartialEq, Eq)] #[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Status { pub enum Status {
@@ -23,22 +23,20 @@ pub struct StatusEntry {
pub reason: Option<CrlReason>, pub reason: Option<CrlReason>,
} }
#[derive(Debug, Clone)] #[derive(Debug, Clone, Default)]
pub struct Snapshot { pub struct Snapshot {
pub generated_at: DateTime<Utc>,
pub entries: HashMap<Vec<u8>, StatusEntry>, pub entries: HashMap<Vec<u8>, StatusEntry>,
} }
#[derive(Debug, Deserialize)] #[derive(Debug, Deserialize, Serialize)]
struct RawSnapshot { struct RawSnapshot {
schema: String, schema: String,
issuer: RawIssuer, issuer: RawIssuer,
generated_at: DateTime<Utc>,
#[serde(default)] #[serde(default)]
entries: Vec<RawEntry>, entries: Vec<RawEntry>,
} }
#[derive(Debug, Deserialize)] #[derive(Debug, Deserialize, Serialize)]
struct RawIssuer { struct RawIssuer {
#[serde(default)] #[serde(default)]
id: String, id: String,
@@ -46,7 +44,7 @@ struct RawIssuer {
sha256: String, sha256: String,
} }
#[derive(Debug, Deserialize)] #[derive(Debug, Deserialize, Serialize)]
struct RawEntry { struct RawEntry {
serial_hex: String, serial_hex: String,
status: String, status: String,
@@ -63,13 +61,12 @@ impl Snapshot {
expected_sha256: &str, expected_sha256: &str,
expected_ca_id: &str, expected_ca_id: &str,
) -> Result<Self, SnapshotError> { ) -> Result<Self, SnapshotError> {
let raw: RawSnapshot = serde_json::from_slice(bytes)?; cbor2::validate_slice(bytes).map_err(|error| SnapshotError::Cbor(error.to_string()))?;
let raw: RawSnapshot =
cbor2::from_slice(bytes).map_err(|error| SnapshotError::Cbor(error.to_string()))?;
if raw.schema != SCHEMA { if raw.schema != SCHEMA {
return Err(SnapshotError::Schema(raw.schema)); return Err(SnapshotError::Schema(raw.schema));
} }
if raw.generated_at > Utc::now() {
return Err(SnapshotError::FutureGeneration(raw.generated_at));
}
if !raw.issuer.id.eq_ignore_ascii_case(expected_ca_id) { if !raw.issuer.id.eq_ignore_ascii_case(expected_ca_id) {
return Err(SnapshotError::IssuerIdMismatch { return Err(SnapshotError::IssuerIdMismatch {
expected: expected_ca_id.to_string(), expected: expected_ca_id.to_string(),
@@ -115,10 +112,7 @@ impl Snapshot {
} }
} }
let _ = raw.issuer.sha256; let _ = raw.issuer.sha256;
Ok(Self { Ok(Self { entries })
generated_at: raw.generated_at,
entries,
})
} }
pub fn status_for_serial(&self, serial: &[u8]) -> Option<&StatusEntry> { pub fn status_for_serial(&self, serial: &[u8]) -> Option<&StatusEntry> {
@@ -174,12 +168,10 @@ pub fn hash_sha256(bytes: &[u8]) -> Vec<u8> {
#[derive(Debug, thiserror::Error)] #[derive(Debug, thiserror::Error)]
pub enum SnapshotError { pub enum SnapshotError {
#[error("invalid JSON: {0}")] #[error("invalid CBOR: {0}")]
Json(#[from] serde_json::Error), Cbor(String),
#[error("unsupported snapshot schema {0:?}")] #[error("unsupported snapshot schema {0:?}")]
Schema(String), Schema(String),
#[error("snapshot generation time {0} is in the future")]
FutureGeneration(DateTime<Utc>),
#[error("snapshot issuer ID {actual} does not match configured CA ID {expected}")] #[error("snapshot issuer ID {actual} does not match configured CA ID {expected}")]
IssuerIdMismatch { expected: String, actual: String }, IssuerIdMismatch { expected: String, actual: String },
#[error("snapshot issuer {actual} does not match configured CA {expected}")] #[error("snapshot issuer {actual} does not match configured CA {expected}")]
@@ -194,28 +186,64 @@ pub enum SnapshotError {
DuplicateSerial(String), DuplicateSerial(String),
} }
#[cfg(test)]
pub(crate) fn test_body(sha: &str) -> Vec<u8> {
let raw = RawSnapshot {
schema: SCHEMA.to_string(),
issuer: RawIssuer {
id: "ca".to_string(),
sha256: sha.to_string(),
},
entries: vec![RawEntry {
serial_hex: "01".to_string(),
status: "valid".to_string(),
revoked_at: None,
reason: None,
}],
};
cbor2::to_vec(&raw).unwrap()
}
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
fn body(sha: &str) -> String { fn raw_body(sha: &str, schema: &str) -> RawSnapshot {
format!( RawSnapshot {
r#"{{ schema: schema.to_string(),
"schema": "kca-ocsp-status/v1", issuer: RawIssuer {
"issuer": {{"id": "ca", "sha256": "{sha}", "subject": "CN=CA"}}, id: "ca".to_string(),
"generated_at": "2026-01-01T00:00:00Z", sha256: sha.to_string(),
"entries": [ },
{{"serial_hex": "01ab", "status": "valid", "not_after": "2027-01-01T00:00:00Z"}}, entries: vec![
{{"serial_hex": "00ff", "status": "revoked", "revoked_at": "2026-02-03T04:05:06Z", "reason": "keyCompromise"}} RawEntry {
] serial_hex: "01ab".to_string(),
}}"# status: "valid".to_string(),
) revoked_at: None,
reason: None,
},
RawEntry {
serial_hex: "00ff".to_string(),
status: "revoked".to_string(),
revoked_at: Some(
DateTime::parse_from_rfc3339("2026-02-03T04:05:06Z")
.unwrap()
.with_timezone(&Utc),
),
reason: Some("keyCompromise".to_string()),
},
],
}
}
fn body(sha: &str) -> Vec<u8> {
cbor2::to_vec(&raw_body(sha, SCHEMA)).unwrap()
} }
#[test] #[test]
fn parses_statuses_and_reasons() { fn parses_statuses_and_reasons() {
let sha = sha256_hex(b"ca"); let sha = sha256_hex(b"ca");
let snapshot = Snapshot::parse(body(&sha).as_bytes(), &sha, "ca").unwrap(); let snapshot = Snapshot::parse(&body(&sha), &sha, "ca").unwrap();
let good = snapshot.status_for_serial(&[0x01, 0xab]).unwrap(); let good = snapshot.status_for_serial(&[0x01, 0xab]).unwrap();
assert_eq!(good.status, Status::Good); assert_eq!(good.status, Status::Good);
let revoked = snapshot.status_for_serial(&[0xff]).unwrap(); let revoked = snapshot.status_for_serial(&[0xff]).unwrap();
@@ -227,32 +255,27 @@ mod tests {
#[test] #[test]
fn rejects_wrong_issuer_schema_and_duplicates() { fn rejects_wrong_issuer_schema_and_duplicates() {
let sha = sha256_hex(b"ca"); let sha = sha256_hex(b"ca");
assert!(Snapshot::parse(body(&sha).as_bytes(), "wrong", "ca").is_err()); assert!(Snapshot::parse(&body(&sha), "wrong", "ca").is_err());
assert!(matches!( assert!(matches!(
Snapshot::parse(body(&sha).as_bytes(), &sha, "other"), Snapshot::parse(&body(&sha), &sha, "other"),
Err(SnapshotError::IssuerIdMismatch { .. }) Err(SnapshotError::IssuerIdMismatch { .. })
)); ));
let wrong_schema = body(&sha).replace(SCHEMA, "other"); let wrong_schema = cbor2::to_vec(&raw_body(&sha, "other")).unwrap();
assert!(Snapshot::parse(wrong_schema.as_bytes(), &sha, "ca").is_err()); assert!(Snapshot::parse(&wrong_schema, &sha, "ca").is_err());
let duplicate = body(&sha).replace("\"00ff\"", "\"01ab\""); let mut duplicate = raw_body(&sha, SCHEMA);
assert!(Snapshot::parse(duplicate.as_bytes(), &sha, "ca").is_err()); duplicate.entries[1].serial_hex = duplicate.entries[0].serial_hex.clone();
let duplicate = cbor2::to_vec(&duplicate).unwrap();
assert!(Snapshot::parse(&duplicate, &sha, "ca").is_err());
} }
#[test] #[test]
fn rejects_missing_or_future_generation_time() { fn rejects_legacy_json() {
let sha = sha256_hex(b"ca"); let sha = sha256_hex(b"ca");
let missing = body(&sha) let legacy =
.lines() br#"{"schema":"kca-ocsp-status/v2","issuer":{"id":"ca","sha256":"sha"},"entries":[]}"#;
.filter(|line| !line.contains("\"generated_at\""))
.collect::<Vec<_>>()
.join("\n");
assert!(Snapshot::parse(missing.as_bytes(), &sha, "ca").is_err());
let future = (Utc::now() + chrono::Duration::hours(1)).to_rfc3339();
let future = body(&sha).replace("2026-01-01T00:00:00Z", &future);
assert!(matches!( assert!(matches!(
Snapshot::parse(future.as_bytes(), &sha, "ca"), Snapshot::parse(legacy, &sha, "ca"),
Err(SnapshotError::FutureGeneration(_)) Err(SnapshotError::Cbor(_))
)); ));
} }
+143 -37
View File
@@ -1,64 +1,170 @@
//! S3-compatible object store access. //! Redis OCSP snapshot access.
use std::time::{Duration, SystemTime};
use async_trait::async_trait; use async_trait::async_trait;
use aws_sdk_s3::Client;
use bytes::Bytes; use bytes::Bytes;
use redis::aio::ConnectionManager;
use crate::config::S3Config; use crate::config::RedisConfig;
const FETCH_SNAPSHOT: &str = r#"
local body = redis.call('GET', KEYS[1])
local ttl = redis.call('PTTL', KEYS[1])
return {body, ttl}
"#;
#[derive(Clone)] #[derive(Clone)]
pub struct StatusStore { pub struct StatusStore {
client: Client, connection: ConnectionManager,
bucket: String,
} }
impl StatusStore { impl StatusStore {
pub async fn new(config: &S3Config) -> Self { pub async fn new(config: &RedisConfig) -> Result<Self, StoreError> {
let mut loader = aws_config::defaults(aws_config::BehaviorVersion::latest()); let client = redis::Client::open(config.url.as_str())
if let Some(region) = config.region.as_ref() { .map_err(|error| StoreError::Redis(error.to_string()))?;
loader = loader.region(aws_config::Region::new(region.clone())); let connection = client
} .get_connection_manager()
let shared = loader.load().await; .await
let mut builder = aws_sdk_s3::config::Builder::from(&shared); .map_err(|error| StoreError::Redis(error.to_string()))?;
if let Some(endpoint) = config.endpoint.as_ref() { Ok(Self { connection })
builder = builder.endpoint_url(endpoint);
}
builder = builder.force_path_style(config.force_path_style);
Self {
client: Client::from_conf(builder.build()),
bucket: config.bucket.clone(),
}
} }
} }
#[derive(Debug)]
pub enum StoredSnapshot {
Present { body: Bytes, expires_at: SystemTime },
Missing,
}
#[async_trait] #[async_trait]
pub trait ObjectStore: Send + Sync { pub trait ObjectStore: Send + Sync {
async fn get(&self, key: &str) -> Result<Bytes, StoreError>; async fn get(&self, key: &str) -> Result<StoredSnapshot, StoreError>;
} }
#[async_trait] #[async_trait]
impl ObjectStore for StatusStore { impl ObjectStore for StatusStore {
async fn get(&self, key: &str) -> Result<Bytes, StoreError> { async fn get(&self, key: &str) -> Result<StoredSnapshot, StoreError> {
let response = self let mut connection = self.connection.clone();
.client let (body, pttl): (Option<Vec<u8>>, i64) = redis::Script::new(FETCH_SNAPSHOT)
.get_object()
.bucket(&self.bucket)
.key(key) .key(key)
.send() .invoke_async(&mut connection)
.await .await
.map_err(|error| StoreError::S3(format!("{error:?}")))?; .map_err(|error| StoreError::Redis(error.to_string()))?;
let data = response stored_snapshot(body, pttl, SystemTime::now())
.body }
.collect() }
.await
.map_err(|error| StoreError::S3(format!("{error:?}")))? fn stored_snapshot(
.into_bytes(); body: Option<Vec<u8>>,
Ok(data) pttl: i64,
now: SystemTime,
) -> Result<StoredSnapshot, StoreError> {
match body {
None => Ok(StoredSnapshot::Missing),
Some(_) if pttl == -1 => Err(StoreError::MissingTtl),
Some(_) if pttl <= 0 => Ok(StoredSnapshot::Missing),
Some(body) => {
let expires_at = now
.checked_add(Duration::from_millis(pttl as u64))
.ok_or(StoreError::InvalidTtl(pttl))?;
Ok(StoredSnapshot::Present {
body: Bytes::from(body),
expires_at,
})
}
} }
} }
#[derive(Debug, thiserror::Error)] #[derive(Debug, thiserror::Error)]
pub enum StoreError { pub enum StoreError {
#[error("S3 GetObject failed: {0}")] #[error("Redis request failed: {0}")]
S3(String), Redis(String),
#[error("OCSP snapshot exists without an expiry")]
MissingTtl,
#[error("OCSP snapshot has an invalid TTL: {0}")]
InvalidTtl(i64),
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn missing_and_expired_keys_are_missing_snapshots() {
let now = SystemTime::now();
assert!(matches!(
stored_snapshot(None, -2, now),
Ok(StoredSnapshot::Missing)
));
assert!(matches!(
stored_snapshot(Some(b"stale".to_vec()), 0, now),
Ok(StoredSnapshot::Missing)
));
}
#[test]
fn present_snapshot_uses_pttl_for_expiry() {
let now = SystemTime::now();
let stored = stored_snapshot(Some(b"snapshot".to_vec()), 5_000, now).unwrap();
match stored {
StoredSnapshot::Present { body, expires_at } => {
assert_eq!(body, Bytes::from_static(b"snapshot"));
assert_eq!(expires_at, now + Duration::from_millis(5_000));
}
StoredSnapshot::Missing => panic!("snapshot should be present"),
}
}
#[test]
fn present_snapshot_without_ttl_is_rejected() {
let result = stored_snapshot(Some(b"snapshot".to_vec()), -1, SystemTime::now());
assert!(matches!(result, Err(StoreError::MissingTtl)));
}
#[tokio::test]
#[ignore = "requires KCA_TEST_REDIS_URL"]
async fn real_redis_round_trip() {
let Ok(url) = std::env::var("KCA_TEST_REDIS_URL") else {
return;
};
let config = RedisConfig {
url,
prefix: format!("ocsp:test:{}", std::process::id()),
..Default::default()
};
let key = config.key("ca");
let sha = crate::snapshot::sha256_hex(b"ca");
let store = StatusStore::new(&config).await.unwrap();
let client = redis::Client::open(config.url.as_str()).unwrap();
let mut connection = client.get_connection_manager().await.unwrap();
let _: () = redis::cmd("SET")
.arg(&key)
.arg(crate::snapshot::test_body(&sha))
.arg("EX")
.arg(30)
.query_async(&mut connection)
.await
.unwrap();
let stored = store.get(&key).await.unwrap();
match stored {
StoredSnapshot::Present { body, expires_at } => {
let snapshot = crate::snapshot::Snapshot::parse(&body, &sha, "ca").unwrap();
assert_eq!(snapshot.entries.len(), 1);
assert!(expires_at > SystemTime::now());
}
StoredSnapshot::Missing => panic!("snapshot should be present"),
}
let _: () = redis::cmd("DEL")
.arg(&key)
.query_async(&mut connection)
.await
.unwrap();
assert!(matches!(
store.get(&key).await.unwrap(),
StoredSnapshot::Missing
));
}
} }
+9 -11
View File
@@ -66,19 +66,17 @@ OTEL_EXCLUDED_PATHS=/api/health
#OTEL_METRICS_POLL_INTERVAL_SECONDS=15 #OTEL_METRICS_POLL_INTERVAL_SECONDS=15
# --- OCSP status publisher (optional) -------------------------------------- # --- OCSP status publisher (optional) --------------------------------------
# When a bucket is configured, KCA publishes one full JSON status snapshot # KCA publishes one full CBOR status snapshot per CA. By default it uses the
# per CA. The standalone OCSP responder reads these objects. Uses the normal # same Redis instance and logical database as sessions. Set KCA_OCSP_REDIS_URL
# AWS credential chain (AWS_ACCESS_KEY_ID / # to use a dedicated instance instead (also works when REDIS_URL is unset).
# AWS_SECRET_ACCESS_KEY, role, instance profile, ...). # Keys use a separate prefix and expire unless refreshed by the heartbeat.
#KCA_OCSP_S3_BUCKET=kca-ocsp #KCA_OCSP_REDIS_URL=redis://localhost:6380/0
KCA_OCSP_S3_PREFIX=ocsp KCA_OCSP_REDIS_PREFIX=ocsp
#KCA_OCSP_S3_REGION=eu-central-1
#KCA_OCSP_S3_ENDPOINT=http://localhost:9000
KCA_OCSP_S3_FORCE_PATH_STYLE=false
KCA_OCSP_PUBLISH_INTERVAL_SECONDS=1 KCA_OCSP_PUBLISH_INTERVAL_SECONDS=1
# Republish unchanged snapshots as a liveness heartbeat. Keep this comfortably # Republish unchanged snapshots to renew their Redis TTL. Keep the heartbeat
# below the responder's s3.stale_grace_seconds (default 60). # comfortably below the key expiry.
KCA_OCSP_HEARTBEAT_INTERVAL_SECONDS=30 KCA_OCSP_HEARTBEAT_INTERVAL_SECONDS=30
KCA_OCSP_REDIS_EXPIRY_SECONDS=60
# --- CORS (optional) ------------------------------------------------------- # --- CORS (optional) -------------------------------------------------------
#CORS_ALLOW_ORIGINS= #CORS_ALLOW_ORIGINS=
+2 -2
View File
@@ -49,11 +49,11 @@ COPY server/uv.lock server/pyproject.toml ./
COPY server/packages/kca-core/pyproject.toml packages/kca-core/ COPY server/packages/kca-core/pyproject.toml packages/kca-core/
COPY server/packages/kca-app/pyproject.toml packages/kca-app/ COPY server/packages/kca-app/pyproject.toml packages/kca-app/
RUN --mount=type=cache,target=/root/.cache/uv \ RUN --mount=type=cache,target=/root/.cache/uv \
uv sync --frozen --no-install-workspace --package kca-app --extra otel --extra s3 --no-dev uv sync --frozen --no-install-workspace --package kca-app --extra otel --no-dev
COPY server/packages/ ./packages/ COPY server/packages/ ./packages/
RUN --mount=type=cache,target=/root/.cache/uv \ RUN --mount=type=cache,target=/root/.cache/uv \
uv sync --locked --no-editable --package kca-app --extra otel --extra s3 --no-dev \ uv sync --locked --no-editable --package kca-app --extra otel --no-dev \
--reinstall-package kca-core --reinstall-package kca-app --reinstall-package kca-core --reinstall-package kca-app
COPY server/migrations ./migrations COPY server/migrations ./migrations
+1 -3
View File
@@ -26,14 +26,12 @@ dependencies = [
"PyYAML", "PyYAML",
"redis", "redis",
"cryptography>=44", "cryptography>=44",
"cbor2",
"pyOpenSSL>=25", "pyOpenSSL>=25",
"paramiko>=3.5", "paramiko>=3.5",
] ]
[project.optional-dependencies] [project.optional-dependencies]
s3 = [
"aioboto3>=13.0",
]
otel = [ otel = [
"kaya-otel>=0.0.7", "kaya-otel>=0.0.7",
"opentelemetry-instrumentation-asyncpg>=0.66b0", "opentelemetry-instrumentation-asyncpg>=0.66b0",
+23 -13
View File
@@ -167,6 +167,7 @@ class Settings:
app_host: str app_host: str
app_port: int app_port: int
redis_url: Optional[str] redis_url: Optional[str]
ocsp_redis_url: Optional[str]
static_dir: str static_dir: str
logging_config: Optional[str] logging_config: Optional[str]
#: AES-256-GCM master key protecting private keys at rest. #: AES-256-GCM master key protecting private keys at rest.
@@ -190,19 +191,32 @@ class Settings:
otel_excluded_paths: Tuple[str, ...] otel_excluded_paths: Tuple[str, ...]
otel_metric_export_interval_millis: int otel_metric_export_interval_millis: int
otel_metrics_poll_interval_seconds: int otel_metrics_poll_interval_seconds: int
ocsp_s3_bucket: Optional[str] ocsp_redis_prefix: str
ocsp_s3_prefix: str
ocsp_s3_region: Optional[str]
ocsp_s3_endpoint: Optional[str]
ocsp_s3_force_path_style: bool
ocsp_publish_interval_seconds: int ocsp_publish_interval_seconds: int
ocsp_heartbeat_interval_seconds: int ocsp_heartbeat_interval_seconds: int
ocsp_redis_expiry_seconds: int
@property
def ocsp_effective_redis_url(self) -> Optional[str]:
"""Dedicated OCSP Redis when configured, otherwise the session Redis."""
return self.ocsp_redis_url or self.redis_url
@staticmethod @staticmethod
def from_env() -> "Settings": def from_env() -> "Settings":
raw_master = os.environ.get("KCA_MASTER_KEY") raw_master = os.environ.get("KCA_MASTER_KEY")
master_key_is_dev = not raw_master master_key_is_dev = not raw_master
master_key = decode_master_key(raw_master) if raw_master else _DEV_MASTER_KEY master_key = decode_master_key(raw_master) if raw_master else _DEV_MASTER_KEY
ocsp_heartbeat_interval_seconds = max(
1, int(_env("KCA_OCSP_HEARTBEAT_INTERVAL_SECONDS", "30"))
)
ocsp_redis_expiry_seconds = max(
1, int(_env("KCA_OCSP_REDIS_EXPIRY_SECONDS", "60"))
)
if ocsp_redis_expiry_seconds <= ocsp_heartbeat_interval_seconds:
raise RuntimeError(
"KCA_OCSP_REDIS_EXPIRY_SECONDS must be greater than "
"KCA_OCSP_HEARTBEAT_INTERVAL_SECONDS"
)
return Settings( return Settings(
database_url=os.environ.get("DATABASE_URL") database_url=os.environ.get("DATABASE_URL")
or _database_url_from_parts( or _database_url_from_parts(
@@ -218,6 +232,7 @@ class Settings:
app_host=_env("APP_HOST", "0.0.0.0"), app_host=_env("APP_HOST", "0.0.0.0"),
app_port=int(_env("APP_PORT", "8080")), app_port=int(_env("APP_PORT", "8080")),
redis_url=os.environ.get("REDIS_URL") or None, redis_url=os.environ.get("REDIS_URL") or None,
ocsp_redis_url=os.environ.get("KCA_OCSP_REDIS_URL") or None,
static_dir=_env("STATIC_DIR", "web/dist"), static_dir=_env("STATIC_DIR", "web/dist"),
logging_config=os.environ.get("LOGGING_CONFIG") or None, logging_config=os.environ.get("LOGGING_CONFIG") or None,
master_key=master_key, master_key=master_key,
@@ -243,17 +258,12 @@ class Settings:
otel_metrics_poll_interval_seconds=int( otel_metrics_poll_interval_seconds=int(
_env("OTEL_METRICS_POLL_INTERVAL_SECONDS", "15") _env("OTEL_METRICS_POLL_INTERVAL_SECONDS", "15")
), ),
ocsp_s3_bucket=os.environ.get("KCA_OCSP_S3_BUCKET") or None, ocsp_redis_prefix=_env("KCA_OCSP_REDIS_PREFIX", "ocsp").strip("/:"),
ocsp_s3_prefix=_env("KCA_OCSP_S3_PREFIX", "ocsp").strip("/"),
ocsp_s3_region=os.environ.get("KCA_OCSP_S3_REGION") or None,
ocsp_s3_endpoint=os.environ.get("KCA_OCSP_S3_ENDPOINT") or None,
ocsp_s3_force_path_style=_env_bool("KCA_OCSP_S3_FORCE_PATH_STYLE"),
ocsp_publish_interval_seconds=max( ocsp_publish_interval_seconds=max(
1, int(_env("KCA_OCSP_PUBLISH_INTERVAL_SECONDS", "1")) 1, int(_env("KCA_OCSP_PUBLISH_INTERVAL_SECONDS", "1"))
), ),
ocsp_heartbeat_interval_seconds=max( ocsp_heartbeat_interval_seconds=ocsp_heartbeat_interval_seconds,
1, int(_env("KCA_OCSP_HEARTBEAT_INTERVAL_SECONDS", "30")) ocsp_redis_expiry_seconds=ocsp_redis_expiry_seconds,
),
) )
+56 -62
View File
@@ -1,20 +1,20 @@
"""Per-CA OCSP status snapshots and S3 publication. """Per-CA OCSP status snapshots and Redis publication.
KCA is the source of truth for certificate status. This module builds a KCA is the source of truth for certificate status. This module builds a
compact JSON document per CA and uploads it to an S3-compatible bucket so compact CBOR document per CA and stores it in the selected Redis instance, so
that the standalone OCSP responder never needs database access or a CA the standalone OCSP responder never needs database access or a CA private key.
private key.
""" """
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
import json
import logging import logging
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
from typing import Any, Awaitable, Callable, Dict, Optional from typing import Any, Awaitable, Callable, Dict, Optional
from uuid import UUID from uuid import UUID
import cbor2
from kaya.core import KayaApp, KayaMixin from kaya.core import KayaApp, KayaMixin
from redis.asyncio import Redis
from tortoise.expressions import Q from tortoise.expressions import Q
from .config import Settings, settings from .config import Settings, settings
@@ -22,7 +22,7 @@ from .models import Certificate, CertificateRevocation, OcspSnapshot, Vault
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
SCHEMA = "kca-ocsp-status/v1" SCHEMA = "kca-ocsp-status/v2"
def _iso(value: Optional[datetime]) -> Optional[str]: def _iso(value: Optional[datetime]) -> Optional[str]:
@@ -33,11 +33,11 @@ def _iso(value: Optional[datetime]) -> Optional[str]:
return value.astimezone(timezone.utc).isoformat() return value.astimezone(timezone.utc).isoformat()
def object_key(ca_id: str) -> str: def redis_key(ca_id: str) -> str:
"""The S3 key for one CA database row; the responder uses the same CA ID.""" """The Redis key for one CA database row; the responder uses the same key."""
prefix = settings.ocsp_s3_prefix.strip("/") prefix = settings.ocsp_redis_prefix.strip("/:")
name = f"{ca_id.lower()}.json" name = ca_id.lower()
return f"{prefix}/{name}" if prefix else name return f"{prefix}:{name}" if prefix else name
async def snapshot_for_ca(ca: Certificate) -> Dict[str, Any]: async def snapshot_for_ca(ca: Certificate) -> Dict[str, Any]:
@@ -73,7 +73,6 @@ async def snapshot_for_ca(ca: Certificate) -> Dict[str, Any]:
"sha256": ca.fingerprint_sha256.lower(), "sha256": ca.fingerprint_sha256.lower(),
"subject": cert_subject(ca), "subject": cert_subject(ca),
}, },
"generated_at": _iso(datetime.now(timezone.utc)),
"entries": entries, "entries": entries,
} }
@@ -84,59 +83,49 @@ def cert_subject(ca: Certificate) -> str:
return cert_core.parse_certificate(bytes(ca.cert_der)).subject.rfc4514_string() return cert_core.parse_certificate(bytes(ca.cert_der)).subject.rfc4514_string()
def _s3_client() -> Any: async def publish_snapshot(
import aioboto3 ca: Certificate,
from botocore.config import Config snapshot: Optional[Dict[str, Any]] = None,
client: Optional[Redis] = None,
kwargs: Dict[str, Any] = { ) -> str:
"region_name": settings.ocsp_s3_region, """Store one CA snapshot with an expiry and return its Redis key."""
"endpoint_url": settings.ocsp_s3_endpoint,
}
if settings.ocsp_s3_force_path_style:
kwargs["config"] = Config(s3={"addressing_style": "path"})
# Only pass region and endpoint values that are explicitly configured.
kwargs = {key: value for key, value in kwargs.items() if value is not None}
return aioboto3.Session().client("s3", **kwargs)
async def publish_snapshot(ca: Certificate, snapshot: Optional[Dict[str, Any]] = None) -> str:
"""Upload one CA snapshot and return its S3 key."""
if not settings.ocsp_s3_bucket:
raise RuntimeError("KCA_OCSP_S3_BUCKET is not configured")
if snapshot is None: if snapshot is None:
snapshot = await snapshot_for_ca(ca) snapshot = await snapshot_for_ca(ca)
body = json.dumps(snapshot, indent=2, sort_keys=True).encode("utf-8") body = cbor2.dumps(snapshot)
key = object_key(str(ca.id)) key = redis_key(str(ca.id))
async with _s3_client() as client: owned_client = None
await client.put_object( if client is None:
Bucket=settings.ocsp_s3_bucket, if not settings.ocsp_effective_redis_url:
Key=key, raise RuntimeError("REDIS_URL is not configured")
Body=body, client = Redis.from_url(settings.ocsp_effective_redis_url)
ContentType="application/json", owned_client = client
CacheControl="no-store", try:
) await client.set(key, body, ex=settings.ocsp_redis_expiry_seconds)
finally:
if owned_client is not None:
await owned_client.aclose()
return key return key
async def mark_ocsp_dirty(ca_id: UUID) -> None: async def mark_ocsp_dirty(ca_id: UUID) -> None:
"""Mark one issuing CA's status object as needing a republish.""" """Mark one issuing CA's status snapshot as needing a republish."""
if not settings.ocsp_s3_bucket: if not settings.ocsp_effective_redis_url:
return return
row = await OcspSnapshot.get_or_none(ca_id=ca_id) row = await OcspSnapshot.get_or_none(ca_id=ca_id)
if row is None: if row is None:
await OcspSnapshot.create(ca_id=ca_id, dirty=True) await OcspSnapshot.create(ca_id=ca_id, dirty=True)
return return
# Always save, even when already dirty: updated_at is the publisher's # Always save, even when already dirty: updated_at is the publisher's
# optimistic-concurrency token for changes made during an upload. # optimistic-concurrency token for changes made during a write.
row.dirty = True row.dirty = True
row.last_error = None # type: ignore[assignment] row.last_error = None # type: ignore[assignment]
await row.save(update_fields=["dirty", "last_error", "updated_at"]) await row.save(update_fields=["dirty", "last_error", "updated_at"])
async def mark_all_cas_dirty(vault: Vault) -> None: async def mark_all_cas_dirty(vault: Vault) -> None:
if not settings.ocsp_s3_bucket: if not settings.ocsp_effective_redis_url:
return return
for ca in await Certificate.filter(vault_id=vault.id, is_ca=True): for ca in await Certificate.filter(vault_id=vault.id, is_ca=True):
await mark_ocsp_dirty(ca.id) await mark_ocsp_dirty(ca.id)
@@ -150,10 +139,13 @@ class OcspPublisherMixin(KayaMixin):
interval_seconds: int, interval_seconds: int,
ensure_context: Callable[[], Awaitable[None]], ensure_context: Callable[[], Awaitable[None]],
heartbeat_interval_seconds: int = 30, heartbeat_interval_seconds: int = 30,
redis_client: Optional[Redis] = None,
) -> None: ) -> None:
self._interval = max(1, interval_seconds) self._interval = max(1, interval_seconds)
self._heartbeat_interval = max(1, heartbeat_interval_seconds) self._heartbeat_interval = max(1, heartbeat_interval_seconds)
self._ensure_context = ensure_context self._ensure_context = ensure_context
self._redis_client = redis_client
self._owns_redis_client = False
self._task: Optional[asyncio.Task] = None self._task: Optional[asyncio.Task] = None
self._bootstrapped = False self._bootstrapped = False
@@ -162,21 +154,19 @@ class OcspPublisherMixin(KayaMixin):
return None return None
def setup(self, loop: asyncio.AbstractEventLoop) -> None: def setup(self, loop: asyncio.AbstractEventLoop) -> None:
if not settings.ocsp_s3_bucket: if not settings.ocsp_effective_redis_url:
log.info("OCSP S3 publication disabled (KCA_OCSP_S3_BUCKET unset)") log.info(
"OCSP Redis publication disabled "
"(REDIS_URL and KCA_OCSP_REDIS_URL unset)"
)
return return
try: if self._redis_client is None:
import aioboto3 # noqa: F401 self._redis_client = Redis.from_url(settings.ocsp_effective_redis_url)
except ImportError as exc: self._owns_redis_client = True
raise RuntimeError(
"KCA_OCSP_S3_BUCKET is set but aioboto3 is not installed; "
"install kca-app with the 's3' extra"
) from exc
self._task = loop.create_task(self._poll()) self._task = loop.create_task(self._poll())
log.info( log.info(
"OCSP S3 publication enabled: bucket=%s prefix=%s interval=%ss", "OCSP Redis publication enabled: prefix=%s interval=%ss",
settings.ocsp_s3_bucket, settings.ocsp_redis_prefix or "(root)",
settings.ocsp_s3_prefix or "(root)",
self._interval, self._interval,
) )
@@ -184,6 +174,10 @@ class OcspPublisherMixin(KayaMixin):
if self._task is not None: if self._task is not None:
self._task.cancel() self._task.cancel()
self._task = None self._task = None
if self._owns_redis_client and self._redis_client is not None:
loop.create_task(self._redis_client.aclose())
self._redis_client = None
self._owns_redis_client = False
async def _poll(self) -> None: async def _poll(self) -> None:
# Let the Tortoise mixin establish its database context before the # Let the Tortoise mixin establish its database context before the
@@ -193,7 +187,7 @@ class OcspPublisherMixin(KayaMixin):
try: try:
if not self._bootstrapped: if not self._bootstrapped:
# Existing CAs may predate the publisher or have lost # Existing CAs may predate the publisher or have lost
# their S3 objects; publish one snapshot for every CA on # their Redis keys; publish one snapshot for every CA on
# the first successful tick. # the first successful tick.
await self._ensure_context() await self._ensure_context()
for ca in await Certificate.filter(is_ca=True): for ca in await Certificate.filter(is_ca=True):
@@ -227,7 +221,7 @@ class OcspPublisherMixin(KayaMixin):
continue continue
try: try:
snapshot = await snapshot_for_ca(ca) snapshot = await snapshot_for_ca(ca)
key = await publish_snapshot(ca, snapshot) key = await publish_snapshot(ca, snapshot, client=self._redis_client)
published_at = datetime.now(timezone.utc) published_at = datetime.now(timezone.utc)
cleared = await OcspSnapshot.filter( cleared = await OcspSnapshot.filter(
ca_id=row.ca_id, ca_id=row.ca_id,
@@ -257,7 +251,7 @@ def publisher_mixin_from_settings(
settings_obj: Settings, settings_obj: Settings,
ensure_context: Callable[[], Awaitable[None]], ensure_context: Callable[[], Awaitable[None]],
) -> Optional[OcspPublisherMixin]: ) -> Optional[OcspPublisherMixin]:
if not settings_obj.ocsp_s3_bucket: if not settings_obj.ocsp_effective_redis_url:
return None return None
return OcspPublisherMixin( return OcspPublisherMixin(
settings_obj.ocsp_publish_interval_seconds, settings_obj.ocsp_publish_interval_seconds,
@@ -271,8 +265,8 @@ __all__ = [
"SCHEMA", "SCHEMA",
"mark_all_cas_dirty", "mark_all_cas_dirty",
"mark_ocsp_dirty", "mark_ocsp_dirty",
"object_key",
"publish_snapshot", "publish_snapshot",
"publisher_mixin_from_settings", "publisher_mixin_from_settings",
"redis_key",
"snapshot_for_ca", "snapshot_for_ca",
] ]
@@ -155,10 +155,9 @@ def register(app: KayaApp, kca: "Kca") -> None:
row = await OcspSnapshot.get_or_none(ca_id=cert.id) row = await OcspSnapshot.get_or_none(ca_id=cert.id)
snapshot = await snapshot_for_ca(cert) snapshot = await snapshot_for_ca(cert)
await send_json(ctx, 200, { await send_json(ctx, 200, {
"enabled": bool(settings.ocsp_s3_bucket), "enabled": bool(settings.ocsp_effective_redis_url),
"bucket": settings.ocsp_s3_bucket, "prefix": settings.ocsp_redis_prefix,
"prefix": settings.ocsp_s3_prefix, "dirty": row.dirty if row else bool(settings.ocsp_effective_redis_url),
"dirty": row.dirty if row else bool(settings.ocsp_s3_bucket),
"published_at": row.published_at.isoformat() if row and row.published_at else None, "published_at": row.published_at.isoformat() if row and row.published_at else None,
"last_error": row.last_error if row else None, "last_error": row.last_error if row else None,
"snapshot": snapshot, "snapshot": snapshot,
+71
View File
@@ -0,0 +1,71 @@
"""Environment configuration tests."""
from __future__ import annotations
import os
import unittest
from unittest.mock import patch
from kca.config import Settings
class OcspExpirySettingsTest(unittest.TestCase):
def test_defaults(self) -> None:
with patch.dict(os.environ, {}, clear=True):
settings = Settings.from_env()
self.assertEqual(30, settings.ocsp_heartbeat_interval_seconds)
self.assertEqual(60, settings.ocsp_redis_expiry_seconds)
def test_expiry_must_exceed_heartbeat(self) -> None:
with patch.dict(
os.environ,
{
"KCA_OCSP_HEARTBEAT_INTERVAL_SECONDS": "30",
"KCA_OCSP_REDIS_EXPIRY_SECONDS": "30",
},
clear=True,
):
with self.assertRaisesRegex(RuntimeError, "must be greater"):
Settings.from_env()
class OcspRedisSettingsTest(unittest.TestCase):
def test_defaults_to_disabled(self) -> None:
with patch.dict(os.environ, {}, clear=True):
settings = Settings.from_env()
self.assertIsNone(settings.redis_url)
self.assertIsNone(settings.ocsp_redis_url)
self.assertIsNone(settings.ocsp_effective_redis_url)
def test_falls_back_to_session_redis(self) -> None:
with patch.dict(
os.environ, {"REDIS_URL": "redis://sessions:6379/0"}, clear=True
):
settings = Settings.from_env()
self.assertEqual(
"redis://sessions:6379/0", settings.ocsp_effective_redis_url
)
def test_uses_dedicated_ocsp_redis(self) -> None:
with patch.dict(
os.environ,
{
"REDIS_URL": "redis://sessions:6379/0",
"KCA_OCSP_REDIS_URL": "redis://ocsp:6379/0",
},
clear=True,
):
settings = Settings.from_env()
self.assertEqual("redis://sessions:6379/0", settings.redis_url)
self.assertEqual("redis://ocsp:6379/0", settings.ocsp_effective_redis_url)
def test_dedicated_ocsp_redis_without_session_redis(self) -> None:
with patch.dict(
os.environ, {"KCA_OCSP_REDIS_URL": "redis://ocsp:6379/0"}, clear=True
):
settings = Settings.from_env()
self.assertIsNone(settings.redis_url)
self.assertEqual("redis://ocsp:6379/0", settings.ocsp_effective_redis_url)
if __name__ == "__main__":
unittest.main()
+41 -50
View File
@@ -941,7 +941,8 @@ class OcspSnapshotTests(unittest.TestCase):
payload = response.json() payload = response.json()
assert payload["enabled"] is False assert payload["enabled"] is False
snapshot = payload["snapshot"] snapshot = payload["snapshot"]
assert snapshot["schema"] == "kca-ocsp-status/v1" assert snapshot["schema"] == "kca-ocsp-status/v2"
assert "generated_at" not in snapshot
assert snapshot["issuer"]["sha256"] == ca["fingerprint_sha256"] assert snapshot["issuer"]["sha256"] == ca["fingerprint_sha256"]
entries = {entry["serial_hex"]: entry for entry in snapshot["entries"]} entries = {entry["serial_hex"]: entry for entry in snapshot["entries"]}
assert len(snapshot["entries"]) == 1 assert len(snapshot["entries"]) == 1
@@ -1072,7 +1073,7 @@ class OcspSnapshotTests(unittest.TestCase):
ca = await _cert(client, vault_id, ca_key["id"], "Dirty CA", ca=True) ca = await _cert(client, vault_id, ca_key["id"], "Dirty CA", ca=True)
leaf_key = await _key(client, vault_id, "dirty leaf") leaf_key = await _key(client, vault_id, "dirty leaf")
leaf = await _cert(client, vault_id, leaf_key["id"], "Dirty Leaf", signer_id=ca["id"]) leaf = await _cert(client, vault_id, leaf_key["id"], "Dirty Leaf", signer_id=ca["id"])
configured = replace(ocsp.settings, ocsp_s3_bucket="test-bucket") configured = replace(ocsp.settings, redis_url="redis://localhost:6379/0")
with patch.object(ocsp, "settings", configured): with patch.object(ocsp, "settings", configured):
response = await client.post( response = await client.post(
f"/api/vaults/{vault_id}/certificates/{leaf['id']}/revoke", f"/api/vaults/{vault_id}/certificates/{leaf['id']}/revoke",
@@ -1099,7 +1100,7 @@ class OcspSnapshotTests(unittest.TestCase):
leaf = await _cert( leaf = await _cert(
client, vault_id, leaf_key["id"], "Existing Dirty Leaf", signer_id=ca["id"] client, vault_id, leaf_key["id"], "Existing Dirty Leaf", signer_id=ca["id"]
) )
configured = replace(ocsp.settings, ocsp_s3_bucket="test-bucket") configured = replace(ocsp.settings, redis_url="redis://localhost:6379/0")
with patch.object(ocsp, "settings", configured): with patch.object(ocsp, "settings", configured):
response = await client.post( response = await client.post(
f"/api/vaults/{vault_id}/certificates/{leaf['id']}/revoke", f"/api/vaults/{vault_id}/certificates/{leaf['id']}/revoke",
@@ -1128,7 +1129,7 @@ class OcspSnapshotTests(unittest.TestCase):
vault_id = await _vault(client) vault_id = await _vault(client)
ca_key = await _key(client, vault_id, "manual dirty ca") ca_key = await _key(client, vault_id, "manual dirty ca")
ca = await _cert(client, vault_id, ca_key["id"], "Manual Dirty CA", ca=True) ca = await _cert(client, vault_id, ca_key["id"], "Manual Dirty CA", ca=True)
configured = replace(ocsp.settings, ocsp_s3_bucket="test-bucket") configured = replace(ocsp.settings, redis_url="redis://localhost:6379/0")
with patch.object(ocsp, "settings", configured): with patch.object(ocsp, "settings", configured):
await OcspSnapshot.create(ca_id=ca["id"], dirty=False) await OcspSnapshot.create(ca_id=ca["id"], dirty=False)
response = await client.post( response = await client.post(
@@ -1176,7 +1177,7 @@ class OcspSnapshotTests(unittest.TestCase):
) )
published = [] published = []
async def fake_publish(publish_ca, snapshot): async def fake_publish(publish_ca, snapshot, client=None):
published.append(snapshot) published.append(snapshot)
if len(published) == 1: if len(published) == 1:
await CertificateRevocation.create( await CertificateRevocation.create(
@@ -1186,9 +1187,9 @@ class OcspSnapshotTests(unittest.TestCase):
reason="keyCompromise", reason="keyCompromise",
) )
await ocsp.mark_ocsp_dirty(publish_ca.id) await ocsp.mark_ocsp_dirty(publish_ca.id)
return ocsp.object_key(str(publish_ca.id)) return ocsp.redis_key(str(publish_ca.id))
configured = replace(ocsp.settings, ocsp_s3_bucket="test-bucket") configured = replace(ocsp.settings, redis_url="redis://localhost:6379/0")
with patch.object(ocsp, "settings", configured), patch.object( with patch.object(ocsp, "settings", configured), patch.object(
ocsp, "publish_snapshot", fake_publish ocsp, "publish_snapshot", fake_publish
): ):
@@ -1215,26 +1216,21 @@ class OcspSnapshotTests(unittest.TestCase):
@async_test @async_test
async def test_publisher_marks_dirty_and_uploads(self) -> None: async def test_publisher_marks_dirty_and_uploads(self) -> None:
import cbor2
from dataclasses import replace from dataclasses import replace
from unittest.mock import patch from unittest.mock import patch
from kca import ocsp from kca import ocsp
from kca.models import OcspSnapshot from kca.models import OcspSnapshot
class FakeS3: class FakeRedis:
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc, traceback):
return None
def __init__(self): def __init__(self):
self.calls = [] self.calls = []
async def put_object(self, **kwargs): async def set(self, name, value, ex=None):
self.calls.append(kwargs) self.calls.append({"name": name, "value": value, "ex": ex})
fake = FakeS3() fake = FakeRedis()
async with api_client() as client: async with api_client() as client:
with oidc_user("alice"): with oidc_user("alice"):
vault_id = await _vault(client) vault_id = await _vault(client)
@@ -1242,25 +1238,25 @@ class OcspSnapshotTests(unittest.TestCase):
ca = await _cert(client, vault_id, ca_key["id"], "OCSP Publish CA", ca=True) ca = await _cert(client, vault_id, ca_key["id"], "OCSP Publish CA", ca=True)
configured = replace( configured = replace(
ocsp.settings, ocsp.settings,
ocsp_s3_bucket="test-bucket", redis_url="redis://localhost:6379/0",
ocsp_s3_prefix="ocsp", ocsp_redis_prefix="ocsp",
) )
with patch.object(ocsp, "settings", configured), patch.object( with patch.object(ocsp, "settings", configured):
ocsp, "_s3_client", lambda: fake
):
await ocsp.mark_ocsp_dirty(ca["id"]) await ocsp.mark_ocsp_dirty(ca["id"])
row = await OcspSnapshot.get(ca_id=ca["id"]) row = await OcspSnapshot.get(ca_id=ca["id"])
assert row.dirty is True assert row.dirty is True
async def ensure_context(): async def ensure_context():
return None return None
mixin = ocsp.OcspPublisherMixin(1, ensure_context) mixin = ocsp.OcspPublisherMixin(1, ensure_context, redis_client=fake)
await mixin.run_once() await mixin.run_once()
assert len(fake.calls) == 1 assert len(fake.calls) == 1
call = fake.calls[0] call = fake.calls[0]
assert call["Bucket"] == "test-bucket" assert call["name"] == f"ocsp:{ca['id']}"
assert call["Key"] == f"ocsp/{ca['id']}.json" published = cbor2.loads(call["value"])
assert b"kca-ocsp-status/v1" in call["Body"] assert published["schema"] == "kca-ocsp-status/v2"
assert "generated_at" not in published
assert call["ex"] == 60
row = await OcspSnapshot.get(ca_id=ca["id"]) row = await OcspSnapshot.get(ca_id=ca["id"])
assert row.dirty is False assert row.dirty is False
assert row.published_at is not None assert row.published_at is not None
@@ -1285,7 +1281,7 @@ class OcspSnapshotTests(unittest.TestCase):
) )
published = [] published = []
async def fake_publish(publish_ca, snapshot): async def fake_publish(publish_ca, snapshot, client=None):
published.append(snapshot) published.append(snapshot)
return "key" return "key"
@@ -1323,7 +1319,7 @@ class OcspSnapshotTests(unittest.TestCase):
) )
published = [] published = []
async def fake_publish(publish_ca, snapshot): async def fake_publish(publish_ca, snapshot, client=None):
published.append(snapshot) published.append(snapshot)
return "key" return "key"
@@ -1340,27 +1336,23 @@ class OcspSnapshotTests(unittest.TestCase):
@async_test @async_test
async def test_imported_ca_cannot_overwrite_other_vault_snapshot(self) -> None: async def test_imported_ca_cannot_overwrite_other_vault_snapshot(self) -> None:
import json import cbor2
from dataclasses import replace from dataclasses import replace
from unittest.mock import patch from unittest.mock import patch
from kca import ocsp from kca import ocsp
from kca.models import Certificate, OcspSnapshot from kca.models import Certificate, OcspSnapshot
class FakeS3: class FakeRedis:
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc, traceback):
return None
def __init__(self): def __init__(self):
self.objects = {} self.objects = {}
self.expiries = {}
async def put_object(self, **kwargs): async def set(self, name, value, ex=None):
self.objects[kwargs["Key"]] = kwargs["Body"] self.objects[name] = value
self.expiries[name] = ex
fake = FakeS3() fake = FakeRedis()
async with api_client() as client: async with api_client() as client:
with oidc_user("alice"): with oidc_user("alice"):
vault_id = await _vault(client) vault_id = await _vault(client)
@@ -1384,14 +1376,12 @@ class OcspSnapshotTests(unittest.TestCase):
configured = replace( configured = replace(
ocsp.settings, ocsp.settings,
ocsp_s3_bucket="test-bucket", redis_url="redis://localhost:6379/0",
ocsp_s3_prefix="ocsp", ocsp_redis_prefix="ocsp",
) )
with patch.object(ocsp, "settings", configured), patch.object( with patch.object(ocsp, "settings", configured):
ocsp, "_s3_client", lambda: fake
):
target = await Certificate.get(id=ca["id"]) target = await Certificate.get(id=ca["id"])
target_key = await ocsp.publish_snapshot(target) target_key = await ocsp.publish_snapshot(target, client=fake)
with oidc_user("bob"): with oidc_user("bob"):
attacker_vault_id = await _vault(client) attacker_vault_id = await _vault(client)
@@ -1419,15 +1409,16 @@ class OcspSnapshotTests(unittest.TestCase):
assert len(response.json()["skipped"]) == 1 assert len(response.json()["skipped"]) == 1
row = await OcspSnapshot.get(ca_id=imported.id) row = await OcspSnapshot.get(ca_id=imported.id)
assert row.dirty is True assert row.dirty is True
imported_key = await ocsp.publish_snapshot(imported) imported_key = await ocsp.publish_snapshot(imported, client=fake)
assert target_key == f"ocsp/{ca['id']}.json" assert target_key == f"ocsp:{ca['id']}"
assert imported_key == f"ocsp/{imported.id}.json" assert imported_key == f"ocsp:{imported.id}"
assert imported_key != target_key assert imported_key != target_key
assert set(fake.objects) == {target_key, imported_key} assert set(fake.objects) == {target_key, imported_key}
target_snapshot = json.loads(fake.objects[target_key]) assert fake.expiries == {target_key: 60, imported_key: 60}
target_snapshot = cbor2.loads(fake.objects[target_key])
target_entries = {entry["serial_hex"]: entry for entry in target_snapshot["entries"]} target_entries = {entry["serial_hex"]: entry for entry in target_snapshot["entries"]}
assert target_entries[leaf["serial_hex"].lower()]["status"] == "revoked" assert target_entries[leaf["serial_hex"].lower()]["status"] == "revoked"
imported_snapshot = json.loads(fake.objects[imported_key]) imported_snapshot = cbor2.loads(fake.objects[imported_key])
imported_serials = {entry["serial_hex"] for entry in imported_snapshot["entries"]} imported_serials = {entry["serial_hex"] for entry in imported_snapshot["entries"]}
assert leaf["serial_hex"].lower() not in imported_serials assert leaf["serial_hex"].lower() not in imported_serials
+161 -994
View File
File diff suppressed because it is too large Load Diff