feat: add aad for crypto (#203)

* feat: add aad for  crypto

* test: update template e2e for dashboard v3

* test: align encrypted fixtures with dashboard v3

* fix: align demo seed provenance with retained v2

Author: hesreallyhim

dashboard_action/runtime/scripts/crypto_artifact.py

@@ -22,11 +22,14 @@
 import storage
 
 
-VERSION = 1
+LEGACY_VERSION = 1
+VERSION = 2
 KDF_ITERATIONS = 600_000
 SALT_BYTES = 16
 IV_BYTES = 12
 ALLOWED_MEMBERS = set(storage.ARTIFACT_FILES)
+RETAINED_ARTIFACT_AAD_LABEL_V2 = "reponomics:retained-artifact:v2:dashboard-data"
+RETAINED_ARTIFACT_AAD_V2 = RETAINED_ARTIFACT_AAD_LABEL_V2.encode("utf-8")
 
 
 def _b64encode(data: bytes) -> str:
@@ -122,13 +125,14 @@ def encrypt(data_dir: Path, output: Path, secret_env: str) -> None:
     iv = os.urandom(IV_BYTES)
     key = _derive_key(secret, salt)
     plaintext = _pack_data_dir(data_dir)
-    ciphertext = AESGCM(key).encrypt(iv, plaintext, None)
+    ciphertext = AESGCM(key).encrypt(iv, plaintext, RETAINED_ARTIFACT_AAD_V2)
     payload = {
         "version": VERSION,
         "created_at": datetime.now(timezone.utc).isoformat(),
         "kdf": "PBKDF2-SHA256",
         "iterations": KDF_ITERATIONS,
         "algorithm": "AES-256-GCM",
+        "aad": RETAINED_ARTIFACT_AAD_LABEL_V2,
         "salt": _b64encode(salt),
         "iv": _b64encode(iv),
         "ciphertext": _b64encode(ciphertext),
@@ -144,19 +148,30 @@ def decrypt(input_path: Path, data_dir: Path, secret_env: str) -> None:
         return
     secret = _load_secret(secret_env)
     payload = json.loads(input_path.read_text(encoding="utf-8"))
-    if payload.get("version") != VERSION:
+    version = payload.get("version")
+    if version not in {LEGACY_VERSION, VERSION}:
         raise ValueError(f"Unsupported encrypted artifact version: {payload.get('version')}")
+    aad = _aad_for_payload(payload)
     key = _derive_key(secret, _b64decode(payload["salt"]))
     plaintext = AESGCM(key).decrypt(
         _b64decode(payload["iv"]),
         _b64decode(payload["ciphertext"]),
-        None,
+        aad,
     )
     _safe_extract(plaintext, data_dir)
     input_path.unlink()
     print(f"Decrypted dashboard data artifact into {data_dir}")
 
 
+def _aad_for_payload(payload: dict[str, object]) -> bytes | None:
+    version = payload.get("version")
+    if version == LEGACY_VERSION:
+        return None
+    if payload.get("aad") != RETAINED_ARTIFACT_AAD_LABEL_V2:
+        raise ValueError("Unsupported encrypted artifact AAD label.")
+    return RETAINED_ARTIFACT_AAD_V2
+
+
 def main() -> None:
     parser = argparse.ArgumentParser()
     subparsers = parser.add_subparsers(dest="command", required=True)

dashboard_action/runtime/scripts/doctor.py

@@ -42,7 +42,10 @@
 )
 from doctor_support import (
     CHUNK_ID_RE,
+    DASHBOARD_CHUNK_AAD_PREFIX,
+    DASHBOARD_SUMMARY_AAD_LABEL,
     ENCRYPTED_DASHBOARD_SCRIPT_ID,
+    EXPORT_AAD_LABEL,
     EXPECTED_DASHBOARD_DATA_VERSION,
     EXPECTED_EXPORT_MANIFEST_VERSION,
     EXPECTED_IV_BYTES,
@@ -80,8 +83,11 @@
 
 __all__ = [
     "CHUNK_ID_RE",
+    "DASHBOARD_CHUNK_AAD_PREFIX",
+    "DASHBOARD_SUMMARY_AAD_LABEL",
     "ENCRYPTED_DASHBOARD_META_NAME",
     "ENCRYPTED_DASHBOARD_SCRIPT_ID",
+    "EXPORT_AAD_LABEL",
     "EXPECTED_DASHBOARD_DATA_VERSION",
     "EXPECTED_EXPORT_MANIFEST_VERSION",
     "EXPECTED_IV_BYTES",

dashboard_action/runtime/scripts/doctor_modules/contracts.py

@@ -6,16 +6,18 @@
 
 from doctor_support import (
     CHUNK_ID_RE,
-    EXPECTED_DASHBOARD_DATA_VERSION,
+    EXPECTED_ENCRYPTED_DASHBOARD_DATA_VERSION,
     EXPECTED_KDF_HASH,
     EXPECTED_KDF_ITERATIONS,
     EXPECTED_KDF_NAME,
+    EXPECTED_PLAINTEXT_DASHBOARD_DATA_VERSION,
     EXPECTED_SALT_BYTES,
     DashboardDoctorError as _DashboardDoctorError,
     DetectedDashboardMode,
     DoctorDataMode,
     DoctorStage,
     _b64_decode,
+    _dashboard_aad_contract_valid,
     _object_dict,
     _stage,
     _validate_encrypted_blob_token,
@@ -208,7 +210,7 @@ def _encrypted_envelope_field_stages(data: dict[str, Any]) -> list[DoctorStage]:
     return [
         _value_stage(
             "browser_envelope_version_valid",
-            data.get("version") == EXPECTED_DASHBOARD_DATA_VERSION,
+            data.get("version") == EXPECTED_ENCRYPTED_DASHBOARD_DATA_VERSION,
             "dashboard data version is supported",
             "dashboard data version is unsupported",
         ),
@@ -224,6 +226,12 @@ def _encrypted_envelope_field_stages(data: dict[str, Any]) -> list[DoctorStage]:
             "encoding is supported",
             "encrypted dashboard encoding is unsupported",
         ),
+        _value_stage(
+            "browser_envelope_aad_valid",
+            _dashboard_aad_contract_valid(data),
+            "AAD contract is supported",
+            "encrypted dashboard AAD contract is unsupported",
+        ),
         _value_stage(
             "browser_envelope_kdf_valid",
             _kdf_contract_valid(data.get("kdf")),
@@ -238,7 +246,7 @@ def _plain_envelope_field_stages(data: dict[str, Any]) -> list[DoctorStage]:
     return [
         _value_stage(
             "browser_envelope_version_valid",
-            data.get("version") == EXPECTED_DASHBOARD_DATA_VERSION,
+            data.get("version") == EXPECTED_PLAINTEXT_DASHBOARD_DATA_VERSION,
             "dashboard data version is supported",
             "dashboard data version is unsupported",
         ),

dashboard_action/runtime/scripts/doctor_modules/data.py

@@ -13,8 +13,10 @@
     _validate_summary_staged,
 )
 from doctor_support import (
+    DASHBOARD_SUMMARY_AAD,
     DoctorSecretResult,
     DoctorStage,
+    _dashboard_chunk_aad,
     _derive_key,
     _object_dict,
     _stage,
@@ -69,6 +71,7 @@ def _diagnose_encrypted_secret(
     summary, summary_stages = _decrypt_gzip_json_staged(
         data.get("summary"),
         key,
+        aad=DASHBOARD_SUMMARY_AAD,
         subject=label,
         auth_stage="summary_authenticates",
         decompress_stage="summary_decompresses",
@@ -152,6 +155,7 @@ def _diagnose_encrypted_chunk(
     chunk, chunk_decode_stages = _decrypt_gzip_json_staged(
         token,
         key,
+        aad=_dashboard_chunk_aad(chunk_id),
         subject=subject,
         auth_stage="chunk_authenticates",
         decompress_stage="chunk_decompresses",

dashboard_action/runtime/scripts/doctor_modules/decode.py

@@ -26,6 +26,7 @@ def _decrypt_gzip_json_staged(
     token: Any,
     key: bytes,
     *,
+    aad: bytes,
     subject: str,
     auth_stage: str,
     decompress_stage: str,
@@ -34,7 +35,7 @@ def _decrypt_gzip_json_staged(
     """Decrypt, decompress, and parse one encrypted gzip+JSON object."""
     stage_names = _DecodeStages(auth=auth_stage, decompress=decompress_stage, json=json_stage)
     stages: list[DoctorStage] = []
-    plaintext, auth_stages = _decrypt_blob_staged(token, key, subject, stage_names)
+    plaintext, auth_stages = _decrypt_blob_staged(token, key, aad, subject, stage_names)
     stages.extend(auth_stages)
     if plaintext is None:
         return None, stages
@@ -52,13 +53,14 @@ def _decrypt_gzip_json_staged(
 def _decrypt_blob_staged(
     token: Any,
     key: bytes,
+    aad: bytes,
     subject: str,
     stage_names: _DecodeStages,
 ) -> tuple[bytes | None, list[DoctorStage]]:
     """Decrypt a token and return authentication stages."""
     try:
         iv, ciphertext = _validate_encrypted_blob_token(token)
-        plaintext = AESGCM(key).decrypt(iv, ciphertext, None)
+        plaintext = AESGCM(key).decrypt(iv, ciphertext, aad)
     except InvalidTag:
         return None, [
             _stage(stage_names.auth, "failed", "AES-GCM authentication failed", subject),

dashboard_action/runtime/scripts/doctor_modules/export_asset.py

@@ -10,6 +10,7 @@
 from cryptography.hazmat.primitives.ciphers.aead import AESGCM
 
 from doctor_support import (
+    EXPORT_AAD,
     DoctorSecretResult,
     DoctorStage,
     DoctorStageStatus,
@@ -135,7 +136,7 @@ def _decrypt_export_with_secret(
     """Decrypt export ciphertext with one accepted dashboard secret."""
     export_key = _derive_key(secret, salt)
     try:
-        plaintext = AESGCM(export_key).decrypt(iv, ciphertext, None)
+        plaintext = AESGCM(export_key).decrypt(iv, ciphertext, EXPORT_AAD)
     except InvalidTag:
         stages.append(_stage("export_decrypts", "failed", "AES-GCM authentication failed", label))
         return None

dashboard_action/runtime/scripts/doctor_modules/handoff.py

@@ -116,6 +116,7 @@ def _ui_handoff_prerequisites(configured_mode: DoctorDataMode, repo_count: int)
     if configured_mode == "encrypted":
         prerequisites.update(
             {
+                "browser_envelope_aad_valid",
                 "browser_envelope_cipher_valid",
                 "browser_envelope_kdf_valid",
                 "browser_envelope_salt_valid",

dashboard_action/runtime/scripts/doctor_modules/manifest.py

@@ -5,6 +5,7 @@
 from typing import Any
 
 from doctor_support import (
+    EXPORT_AAD_LABEL,
     EXPECTED_EXPORT_MANIFEST_VERSION,
     EXPECTED_IV_BYTES,
     EXPECTED_KDF_HASH,
@@ -53,6 +54,8 @@ def _export_manifest_contract_errors(manifest: dict[str, Any]) -> list[str]:
         errors.append("invalid asset path")
     if not isinstance(manifest.get("filename"), str) or not manifest.get("filename"):
         errors.append("missing filename")
+    if manifest.get("aad") != EXPORT_AAD_LABEL:
+        errors.append("unsupported AAD")
     if not _positive_int(manifest.get("ciphertext_size")):
         errors.append("invalid ciphertext size")
     if not _sha256_value_valid(manifest.get("ciphertext_sha256")):

dashboard_action/runtime/scripts/doctor_modules/result.py

@@ -20,6 +20,7 @@
     "browser_envelope_version_valid",
     "browser_envelope_cipher_valid",
     "browser_envelope_kdf_valid",
+    "browser_envelope_aad_valid",
     "browser_envelope_encoding_valid",
     "browser_envelope_salt_valid",
     "browser_envelope_summary_token_valid",

dashboard_action/runtime/scripts/doctor_retained.py

@@ -29,6 +29,10 @@
 
 
 RETAINED_ENCRYPTED_ARTIFACT_NAME = "dashboard-data.enc"
+RETAINED_ARTIFACT_LEGACY_VERSION = 1
+RETAINED_ARTIFACT_VERSION = 2
+RETAINED_ARTIFACT_AAD_LABEL_V2 = "reponomics:retained-artifact:v2:dashboard-data"
+RETAINED_ARTIFACT_AAD_V2 = RETAINED_ARTIFACT_AAD_LABEL_V2.encode("utf-8")
 
 
 def _retained_encrypted_candidates(retained_data_dir: Path | None) -> list[Path]:
@@ -148,17 +152,36 @@ def _object_dict(value: Any) -> dict[str, Any]:
     return {}
 
 
-def _load_retained_encrypted_payload(path: Path) -> tuple[list[DoctorStage], dict[str, Any] | None]:
+def _load_retained_encrypted_payload(
+    path: Path,
+) -> tuple[list[DoctorStage], dict[str, Any] | None]:
     try:
         payload = json.loads(path.read_text(encoding="utf-8"))
     except Exception as exc:
-        return [_stage("retained_artifact_readable", "failed", f"encrypted artifact was not readable JSON: {exc}")], None
+        return [
+            _stage(
+                "retained_artifact_readable",
+                "failed",
+                f"encrypted artifact was not readable JSON: {exc}",
+            )
+        ], None
     if not isinstance(payload, dict):
-        return [_stage("retained_artifact_readable", "failed", "encrypted artifact payload was not a JSON object")], None
+        return [
+            _stage(
+                "retained_artifact_readable",
+                "failed",
+                "encrypted artifact payload was not a JSON object",
+            )
+        ], None
 
     errors: list[str] = []
-    if payload.get("version") != 1:
+    if payload.get("version") not in {RETAINED_ARTIFACT_LEGACY_VERSION, RETAINED_ARTIFACT_VERSION}:
         errors.append("unsupported encrypted artifact version")
+    if (
+        payload.get("version") == RETAINED_ARTIFACT_VERSION
+        and payload.get("aad") != RETAINED_ARTIFACT_AAD_LABEL_V2
+    ):
+        errors.append("unsupported AAD")
     if payload.get("kdf") != "PBKDF2-SHA256":
         errors.append("unsupported KDF")
     if payload.get("iterations") != EXPECTED_KDF_ITERATIONS:
@@ -173,6 +196,12 @@ def _load_retained_encrypted_payload(path: Path) -> tuple[list[DoctorStage], dic
     return [_stage("retained_artifact_readable", "passed", "encrypted artifact payload is readable")], payload
 
 
+def _retained_artifact_aad(payload: dict[str, Any]) -> bytes | None:
+    if payload.get("version") == RETAINED_ARTIFACT_LEGACY_VERSION:
+        return None
+    return RETAINED_ARTIFACT_AAD_V2
+
+
 def _diagnose_encrypted_retained_artifact(
     path: Path,
     *,
@@ -223,7 +252,7 @@ def _diagnose_encrypted_retained_artifact(
     for label, secret in accepted_secrets:
         key = _derive_key(secret, salt)
         try:
-            plaintext = AESGCM(key).decrypt(iv, ciphertext, None)
+            plaintext = AESGCM(key).decrypt(iv, ciphertext, _retained_artifact_aad(payload))
         except InvalidTag:
             stages.append(_stage("retained_artifact_decrypts", "failed", "AES-GCM authentication failed", label))
             continue