fix: attestation templating

api/webhooks.py

@@ -122,48 +122,72 @@ def render_template_string(template, context=None, attestation=None, webhook=Non
             return attestation.get("product_version", "")
         if varname == "attestation_details":
             return attestation
-        # Attestation-derived (load only if needed)
-        if varname in ("raw_attestation", "attestation_json", "attestation_predicate", "attestation_predicate_b64"):
-            if "raw" not in attestation_loaded:
-                attestation_loaded["raw"] = None
-                if storage_path or storage_url:
-                    backend = get_storage_backend()
-                    try:
-                        if storage_path:
-                            attestation_loaded["raw"] = backend.download_file(storage_path)
-                        elif storage_url:
-                            import requests
-                            resp = requests.get(storage_url)
-                            if resp.ok:
-                                attestation_loaded["raw"] = resp.content
-                    except Exception:
-                        attestation_loaded["raw"] = None
-            raw = attestation_loaded["raw"]
-            if raw is None:
+
+        # Only support attestation_raw, attestation_predicate, attestation_predicate_base64
+        if varname in ("attestation_raw", "attestation_predicate", "attestation_predicate_base64"):
+            from models import Attestation as AttestationModel
+            import importlib
+            web_routes = importlib.import_module("web.routes")
+            get_attestation_file_content = getattr(web_routes, "get_attestation_file_content")
+
+            # Try to get Attestation object from DB using id or storage_path
+            att_obj = None
+            att_id = attestation.get("id")
+            storage_path_val = attestation.get("storage_path")
+            if att_id is not None:
+                att_obj = AttestationModel.query.filter_by(id=att_id).first()
+            elif storage_path_val:
+                att_obj = AttestationModel.query.filter_by(storage_path=storage_path_val).first()
+            if not att_obj:
+                return ""
+
+            # --- Get raw file content as in get_attestation_file_content, before decoding ---
+            config = None
+            storage = None
+            try:
+                from flask import current_app
+                config = current_app.config.get('STORAGE_CONFIG')
+                storage = get_storage_backend()
+            except Exception:
+                return ""
+            raw_content = None
+            try:
+                if config.STORAGE_TYPE == 'FILE_MOUNT':
+                    import os
+                    file_path = os.path.join(config.FILE_STORAGE_PATH, att_obj.storage_path)
+                    if not os.path.exists(file_path):
+                        return ""
+                    with open(file_path, "rb") as f:
+                        raw_content = f.read()
+                else:
+                    import urllib.request
+                    url = storage.generate_download_url(att_obj.storage_path)
+                    if not url:
+                        return ""
+                    response = urllib.request.urlopen(url)
+                    raw_content = response.read()
+            except Exception:
+                return ""
+
+            # Now handle the variable
+            if varname == "attestation_raw":
+                return raw_content
+
+            # For predicate and predicate_base64, use the decoded predicate as before
+            predicate, content_type, is_json, error = get_attestation_file_content(att_obj)
+            if error or predicate is None:
                 return ""
-            if varname == "raw_attestation":
-                return raw
-            if varname == "attestation_json":
-                try:
-                    return json.loads(raw)
-                except Exception:
-                    return ""
             if varname == "attestation_predicate":
-                try:
-                    j = json.loads(raw)
-                    return j.get("predicate", "")
-                except Exception:
-                    return ""
-            if varname == "attestation_predicate_b64":
-                try:
-                    import base64
-                    j = json.loads(raw)
-                    pred = j.get("predicate", "")
-                    if isinstance(pred, (dict, list)):
-                        pred = json.dumps(pred)
-                    return base64.b64encode(pred.encode()).decode()
-                except Exception:
-                    return ""
+                return predicate
+            if varname == "attestation_predicate_base64":
+                import base64
+                if isinstance(predicate, (dict, list)):
+                    pred_str = json.dumps(predicate)
+                else:
+                    pred_str = str(predicate)
+                return base64.b64encode(pred_str.encode()).decode()
+            return ""
+
         # Context fields
         if varname.startswith("labels."):
             label_name = varname.split(".", 1)[1]