Publish OCSP snapshots through Redis
This commit is contained in:
Generated
+155
-1044
File diff suppressed because it is too large
Load Diff
@@ -19,7 +19,6 @@ bytes = "1"
|
||||
async-trait = "0.1"
|
||||
base64 = "0.22"
|
||||
hex = "0.4"
|
||||
httpdate = "0.3"
|
||||
sha1 = "0.10"
|
||||
sha2 = "0.10"
|
||||
der = { version = "0.7", features = ["std"] }
|
||||
@@ -31,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"] }
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
+40
-52
@@ -1,58 +1,47 @@
|
||||
# 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.
|
||||
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 same Redis instance and logical database KCA
|
||||
uses for sessions;
|
||||
- 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. KCA needs access to both
|
||||
`ocsp:*` and `kaya:session:*`.
|
||||
|
||||
## 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
|
||||
KCA publishes these keys when `REDIS_URL` is configured. The
|
||||
`kca-ocsp-status/v2` JSON contains all certificates issued by that CA, both
|
||||
valid and revoked, without a generation timestamp. Freshness is carried by the
|
||||
S3 `Expires` header instead. 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.
|
||||
|
||||
## Upgrading from timestamped snapshots
|
||||
|
||||
Version 2 snapshots remove `generated_at` and require an S3 `Expires` header.
|
||||
Upgrade the publisher and responder together: old objects have no `Expires` and
|
||||
are rejected, while old responders cannot parse the version 2 JSON. Until the
|
||||
new publisher writes a snapshot, the new responder returns `tryLater`.
|
||||
Redis key TTL. The responder never infers status from a partial revoked-only
|
||||
list.
|
||||
|
||||
## 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 same Redis URL used by KCA.
|
||||
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:
|
||||
@@ -70,12 +59,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
|
||||
@@ -104,24 +93,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 refresh. Each refresh performs
|
||||
`HeadObject` to obtain `Expires` and the ETag, then `GetObject` with
|
||||
`If-Match` for that ETag. This supports RustFS, which does not return the
|
||||
stored `Expires` value from its local `GetObject` path, and prevents mixing
|
||||
metadata and bodies from different object versions. Concurrent overwrites
|
||||
cause a bounded retry of the complete HEAD/GET sequence.
|
||||
- `Expires` is the hard snapshot expiry. Missing, malformed and expired values
|
||||
are rejected; if no unexpired cached snapshot remains, OCSP returns
|
||||
`tryLater`. The header controls responder freshness only—S3 does not delete
|
||||
the object at that time.
|
||||
- If a refresh fails, a cached snapshot is served only until its own `Expires`
|
||||
timestamp, regardless of how recently it was fetched.
|
||||
- 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) and sets `Expires`
|
||||
using `KCA_OCSP_OBJECT_EXPIRY_SECONDS` (default 60 seconds). Keep the
|
||||
heartbeat comfortably below the object expiry.
|
||||
`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 hard expiry.
|
||||
- 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.
|
||||
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.
|
||||
|
||||
@@ -3,13 +3,11 @@
|
||||
listen: 0.0.0.0:2560
|
||||
next_update_seconds: 10
|
||||
|
||||
s3:
|
||||
bucket: kca-ocsp
|
||||
redis:
|
||||
# Same Redis instance and logical database used by KCA sessions.
|
||||
# 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
|
||||
|
||||
issuers:
|
||||
|
||||
+42
-12
@@ -26,9 +26,9 @@ 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`.
|
||||
///
|
||||
/// A cached snapshot is never used after its S3 `Expires` timestamp,
|
||||
/// A cached snapshot is never used after its Redis key expires,
|
||||
/// including when a refresh fails.
|
||||
pub async fn get<S: ObjectStore + ?Sized>(
|
||||
&self,
|
||||
@@ -60,16 +60,22 @@ impl SnapshotCache {
|
||||
};
|
||||
|
||||
match store.get(key).await {
|
||||
Ok(stored) => {
|
||||
let parsed = validate_expiry(stored.expires_at).and_then(|()| {
|
||||
Snapshot::parse(&stored.body, expected_sha256, expected_ca_id)
|
||||
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: stored.expires_at,
|
||||
expires_at,
|
||||
fetched_at: Instant::now(),
|
||||
};
|
||||
*self.cached.write().await = Some(cached.clone());
|
||||
@@ -109,15 +115,14 @@ fn validate_expiry(expires_at: SystemTime) -> Result<(), String> {
|
||||
if expiry_valid(expires_at) {
|
||||
Ok(())
|
||||
} else {
|
||||
Err(format!(
|
||||
"snapshot expired at {}",
|
||||
httpdate::fmt_http_date(expires_at)
|
||||
))
|
||||
Err(format!("snapshot Redis key expired at {expires_at:?}"))
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum CacheError {
|
||||
#[error("OCSP snapshot is missing")]
|
||||
Missing,
|
||||
#[error("OCSP snapshot unavailable: {0}")]
|
||||
Unavailable(String),
|
||||
}
|
||||
@@ -138,6 +143,7 @@ mod tests {
|
||||
expires_at: SystemTime,
|
||||
calls: AtomicUsize,
|
||||
fail: AtomicBool,
|
||||
missing: AtomicBool,
|
||||
}
|
||||
|
||||
impl FakeStore {
|
||||
@@ -147,6 +153,7 @@ mod tests {
|
||||
expires_at: SystemTime::now() + lifetime,
|
||||
calls: AtomicUsize::new(0),
|
||||
fail: AtomicBool::new(false),
|
||||
missing: AtomicBool::new(false),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -156,6 +163,7 @@ mod tests {
|
||||
expires_at: SystemTime::now() - Duration::from_secs(1),
|
||||
calls: AtomicUsize::new(0),
|
||||
fail: AtomicBool::new(false),
|
||||
missing: AtomicBool::new(false),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -165,9 +173,12 @@ mod tests {
|
||||
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(StoredSnapshot {
|
||||
if self.missing.load(Ordering::SeqCst) {
|
||||
return Ok(StoredSnapshot::Missing);
|
||||
}
|
||||
Ok(StoredSnapshot::Present {
|
||||
body: Bytes::from(self.body.clone()),
|
||||
expires_at: self.expires_at,
|
||||
})
|
||||
@@ -243,4 +254,23 @@ mod tests {
|
||||
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(_))));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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,39 +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,
|
||||
}
|
||||
|
||||
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(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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}")
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -90,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()?;
|
||||
@@ -104,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"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -111,9 +111,9 @@ impl Issuer {
|
||||
pub async fn snapshot<S: crate::store::ObjectStore + ?Sized>(
|
||||
&self,
|
||||
store: &S,
|
||||
config: &crate::config::S3Config,
|
||||
config: &crate::config::RedisConfig,
|
||||
) -> Result<crate::cache::CachedSnapshot, crate::cache::CacheError> {
|
||||
let key = config.object_key(&self.ca_id);
|
||||
let key = config.key(&self.ca_id);
|
||||
self.cache
|
||||
.get(
|
||||
store,
|
||||
|
||||
@@ -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)?;
|
||||
|
||||
@@ -66,14 +66,16 @@ async fn handle_request(state: &AppState, body: &[u8]) -> Response {
|
||||
return ocsp_response(OcspResponse::unauthorized());
|
||||
}
|
||||
|
||||
let cached = match issuer.snapshot(&state.store, &state.config.s3).await {
|
||||
Ok(cached) => cached,
|
||||
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());
|
||||
}
|
||||
};
|
||||
let snapshot = &cached.snapshot;
|
||||
|
||||
let now = SystemTime::now();
|
||||
let produced_at = match OcspGeneralizedTime::try_from(now) {
|
||||
@@ -84,7 +86,7 @@ async fn handle_request(state: &AppState, body: &[u8]) -> Response {
|
||||
}
|
||||
};
|
||||
let (this_update, next_update) =
|
||||
response_update_times(cached.expires_at, state.config.next_update_seconds, now);
|
||||
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) => {
|
||||
@@ -139,13 +141,16 @@ async fn handle_request(state: &AppState, body: &[u8]) -> Response {
|
||||
}
|
||||
|
||||
fn response_update_times(
|
||||
expires_at: SystemTime,
|
||||
expires_at: Option<SystemTime>,
|
||||
next_update_seconds: u64,
|
||||
now: SystemTime,
|
||||
) -> (SystemTime, Option<SystemTime>) {
|
||||
let next_update = now
|
||||
.checked_add(Duration::from_secs(next_update_seconds))
|
||||
.map(|next| next.min(expires_at));
|
||||
.map(|next| match expires_at {
|
||||
Some(expires_at) => next.min(expires_at),
|
||||
None => next,
|
||||
});
|
||||
(now, next_update)
|
||||
}
|
||||
|
||||
@@ -177,12 +182,15 @@ mod tests {
|
||||
let expires_at = base + Duration::from_secs(60);
|
||||
|
||||
let now = base + Duration::from_secs(5);
|
||||
let (this_update, next_update) = response_update_times(expires_at, 10, now);
|
||||
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 = base + Duration::from_secs(55);
|
||||
let (_, next_update) = response_update_times(expires_at, 10, now);
|
||||
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)));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -23,7 +23,7 @@ pub struct StatusEntry {
|
||||
pub reason: Option<CrlReason>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
#[derive(Debug, Clone, Default)]
|
||||
pub struct Snapshot {
|
||||
pub entries: HashMap<Vec<u8>, StatusEntry>,
|
||||
}
|
||||
|
||||
+121
-111
@@ -1,46 +1,40 @@
|
||||
//! S3-compatible object store access.
|
||||
//! Redis OCSP snapshot access.
|
||||
|
||||
use std::time::SystemTime;
|
||||
use std::time::{Duration, SystemTime};
|
||||
|
||||
use async_trait::async_trait;
|
||||
use aws_sdk_s3::error::SdkError;
|
||||
use aws_sdk_s3::operation::get_object::GetObjectError;
|
||||
use aws_sdk_s3::Client;
|
||||
use bytes::Bytes;
|
||||
use redis::aio::ConnectionManager;
|
||||
|
||||
use crate::config::S3Config;
|
||||
use crate::config::RedisConfig;
|
||||
|
||||
const MAX_FETCH_ATTEMPTS: usize = 3;
|
||||
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 struct StoredSnapshot {
|
||||
pub body: Bytes,
|
||||
pub expires_at: SystemTime,
|
||||
pub enum StoredSnapshot {
|
||||
Present { body: Bytes, expires_at: SystemTime },
|
||||
Missing,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
@@ -51,108 +45,124 @@ pub trait ObjectStore: Send + Sync {
|
||||
#[async_trait]
|
||||
impl ObjectStore for StatusStore {
|
||||
async fn get(&self, key: &str) -> Result<StoredSnapshot, StoreError> {
|
||||
// RustFS returns Expires from HeadObject but not GetObject. A
|
||||
// conditional GET binds the body to the metadata returned by HEAD,
|
||||
// even if the publisher overwrites the object between the two calls.
|
||||
for attempt in 1..=MAX_FETCH_ATTEMPTS {
|
||||
let head = self
|
||||
.client
|
||||
.head_object()
|
||||
.bucket(&self.bucket)
|
||||
.key(key)
|
||||
.send()
|
||||
.await
|
||||
.map_err(|error| StoreError::S3(format!("HeadObject failed: {error:?}")))?;
|
||||
let expires_at = parse_expiry(head.expires_string.as_deref())?;
|
||||
let etag = head
|
||||
.e_tag
|
||||
.as_deref()
|
||||
.ok_or(StoreError::MissingEtag)?
|
||||
.to_string();
|
||||
let mut connection = self.connection.clone();
|
||||
let (body, pttl): (Option<Vec<u8>>, i64) = redis::Script::new(FETCH_SNAPSHOT)
|
||||
.key(key)
|
||||
.invoke_async(&mut connection)
|
||||
.await
|
||||
.map_err(|error| StoreError::Redis(error.to_string()))?;
|
||||
stored_snapshot(body, pttl, SystemTime::now())
|
||||
}
|
||||
}
|
||||
|
||||
let response = match self
|
||||
.client
|
||||
.get_object()
|
||||
.bucket(&self.bucket)
|
||||
.key(key)
|
||||
.if_match(etag)
|
||||
.send()
|
||||
.await
|
||||
{
|
||||
Ok(response) => response,
|
||||
Err(error) if attempt < MAX_FETCH_ATTEMPTS && is_precondition_failed(&error) => {
|
||||
continue;
|
||||
}
|
||||
Err(error) => {
|
||||
return Err(StoreError::S3(format!("GetObject failed: {error:?}")));
|
||||
}
|
||||
};
|
||||
let body = response
|
||||
.body
|
||||
.collect()
|
||||
.await
|
||||
.map_err(|error| StoreError::S3(format!("GetObject body failed: {error:?}")))?
|
||||
.into_bytes();
|
||||
return Ok(StoredSnapshot { body, expires_at });
|
||||
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,
|
||||
})
|
||||
}
|
||||
unreachable!("the final GET attempt either returns or fails")
|
||||
}
|
||||
}
|
||||
|
||||
fn parse_expiry(value: Option<&str>) -> Result<SystemTime, StoreError> {
|
||||
let value = value.ok_or(StoreError::MissingExpiry)?;
|
||||
let expires_at = httpdate::parse_http_date(value)
|
||||
.map_err(|_| StoreError::InvalidExpiry(value.to_string()))?;
|
||||
if expires_at <= SystemTime::now() {
|
||||
return Err(StoreError::Expired(value.to_string()));
|
||||
}
|
||||
Ok(expires_at)
|
||||
}
|
||||
|
||||
fn is_precondition_failed(error: &SdkError<GetObjectError>) -> bool {
|
||||
error
|
||||
.as_service_error()
|
||||
.and_then(|error| error.meta().code())
|
||||
== Some("PreconditionFailed")
|
||||
}
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum StoreError {
|
||||
#[error("S3 request failed: {0}")]
|
||||
S3(String),
|
||||
#[error("S3 object has no Expires header")]
|
||||
MissingExpiry,
|
||||
#[error("S3 object has an invalid Expires header: {0}")]
|
||||
InvalidExpiry(String),
|
||||
#[error("S3 object expired at {0}")]
|
||||
Expired(String),
|
||||
#[error("S3 object has no ETag")]
|
||||
MissingEtag,
|
||||
#[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 std::time::Duration;
|
||||
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn parses_future_http_expiry() {
|
||||
let expires = httpdate::fmt_http_date(SystemTime::now() + Duration::from_secs(60));
|
||||
assert!(parse_expiry(Some(&expires)).is_ok());
|
||||
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 rejects_missing_invalid_and_expired_expiry() {
|
||||
assert!(matches!(parse_expiry(None), Err(StoreError::MissingExpiry)));
|
||||
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("round-trip");
|
||||
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("snapshot")
|
||||
.arg("EX")
|
||||
.arg(30)
|
||||
.query_async(&mut connection)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let stored = store.get(&key).await.unwrap();
|
||||
match stored {
|
||||
StoredSnapshot::Present { body, expires_at } => {
|
||||
assert_eq!(body, Bytes::from_static(b"snapshot"));
|
||||
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!(
|
||||
parse_expiry(Some("not-a-date")),
|
||||
Err(StoreError::InvalidExpiry(_))
|
||||
));
|
||||
let expired = httpdate::fmt_http_date(SystemTime::now() - Duration::from_secs(1));
|
||||
assert!(matches!(
|
||||
parse_expiry(Some(&expired)),
|
||||
Err(StoreError::Expired(_))
|
||||
store.get(&key).await.unwrap(),
|
||||
StoredSnapshot::Missing
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user