Use S3 object expiry for OCSP snapshots

server/packages/kca-app/src/kca/config.py

@@ -197,12 +197,24 @@ class Settings:
     ocsp_s3_force_path_style: bool
     ocsp_publish_interval_seconds: int
     ocsp_heartbeat_interval_seconds: int
+    ocsp_object_expiry_seconds: int
 
     @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_object_expiry_seconds = max(
+            1, int(_env("KCA_OCSP_OBJECT_EXPIRY_SECONDS", "60"))
+        )
+        if ocsp_object_expiry_seconds <= ocsp_heartbeat_interval_seconds:
+            raise RuntimeError(
+                "KCA_OCSP_OBJECT_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(
@@ -251,9 +263,8 @@ class Settings:
             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_object_expiry_seconds=ocsp_object_expiry_seconds,
         )
 
 

server/packages/kca-app/src/kca/ocsp.py

@@ -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]:
@@ -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,
     }
 
@@ -115,6 +114,8 @@ async def publish_snapshot(ca: Certificate, snapshot: Optional[Dict[str, Any]] =
             Body=body,
             ContentType="application/json",
             CacheControl="no-store",
+            Expires=datetime.now(timezone.utc)
+            + timedelta(seconds=settings.ocsp_object_expiry_seconds),
         )
 
     return key