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