functioning templating (metadata)

api/webhooks.py

@@ -1,3 +1,19 @@
+ 
+import json
+import io
+from flask import Blueprint, request, jsonify, send_file
+from flask_jwt_extended import get_jwt_identity
+from models import Webhook, User, Attestation
+from app import db
+from auth import jwt_required_with_logging
+import requests
+from datetime import datetime
+from logging_utils import log_event
+from flask import current_app
+
+import re
+from storage import get_storage_backend
+
 def trigger_webhook(webhook, attestation_data, event=None):
     """
     Send a webhook with template support for headers and content.
@@ -7,73 +23,58 @@ def trigger_webhook(webhook, attestation_data, event=None):
     import json, requests
     # headers = {'Content-Type': 'application/json'}
     headers = {}
+    context = attestation_data.get("context", {}) if isinstance(attestation_data, dict) else {}
+    # Render headers with templating
     if webhook.headers:
         try:
             user_headers = json.loads(webhook.headers)
             for k, v in user_headers.items():
-                if isinstance(v, str) and '{secret}' in v:
-                    user_headers[k] = v.replace('{secret}', webhook.secret or '')
+                user_headers[k] = render_template_string(v, context, attestation_data, webhook)
             headers.update(user_headers)
         except Exception as e:
             current_app.logger.error(f"Error parsing webhook headers: {e}")
             return None
-    if webhook.secret and not any('{secret}' in str(v) for v in (json.loads(webhook.headers).values() if webhook.headers else [])):
+    # Render secret header if not templated
+    if webhook.secret and not any("--secret--" in str(v) for v in (json.loads(webhook.headers).values() if webhook.headers else [])):
         headers['X-Webhook-Secret'] = webhook.secret
-    # Prepare content
+    # Prepare content with templating
     if webhook.content:
         try:
             content_str = webhook.content
-            # Replace {attestation_details} with the Python dict (not JSON string)
-            if '{attestation_details}' in content_str:
-                # Safely replace the string with the dict in the parsed object
-                # Parse as JSON, then recursively replace the value
+            # Try to parse as JSON, else treat as string
+            try:
                 obj = json.loads(content_str)
-                def replace_attestation_details(o):
-                    if isinstance(o, dict):
-                        return {k: (attestation_data if v == '{attestation_details}' else replace_attestation_details(v)) for k, v in o.items()}
-                    elif isinstance(o, list):
-                        return [replace_attestation_details(i) for i in o]
-                    elif isinstance(o, str):
-                        if '{event}' in o and event is not None:
-                            o = o.replace('{event}', str(event))
-                        if '{webhook_id}' in o:
-                            o = o.replace('{webhook_id}', str(getattr(webhook, 'id', '')))
-                        return o
-                    else:
-                        return o
-                payload = replace_attestation_details(obj)
-            else:
-                # No {attestation_details}, just replace other fields
-                content_str = content_str
-                if event is not None:
-                    content_str = content_str.replace('{event}', str(event))
-                content_str = content_str.replace('{webhook_id}', str(getattr(webhook, 'id', '')))
-                payload = json.loads(content_str)
+                payload = render_template_string(obj, context, attestation_data, webhook)
+            except Exception:
+                payload = render_template_string(content_str, context, attestation_data, webhook)
+                try:
+                    payload = json.loads(payload)
+                except Exception:
+                    pass
         except Exception as e:
             current_app.logger.error(f"Error parsing webhook content: {e}")
             return None
     else:
         payload = {'attestation': attestation_data}
-    # if event:
-    #     payload['event'] = event
-    # payload['webhook_id'] = webhook.id
+    # Render URL with templating
+    url = render_template_string(webhook.url, context, attestation_data, webhook)
     method = webhook.method or 'POST'
     try:
         req_func = getattr(requests, method.lower(), requests.post)
         # Log sending attempt
         log_event(
             severity=6,  # Informational
-            message=f"Sending webhook to {webhook.url} with payload: {payload}",
+            message=f"Sending webhook to {url} with payload: {payload}",
             event_type=event or getattr(webhook, 'event_type', None),
             user_id=getattr(webhook, 'created_by', None),
             source_ip=None,
             app_name='webhook-sender'
         )
-        resp = req_func(webhook.url, data=json.dumps(payload), headers=headers, timeout=5)
+        resp = req_func(url, data=json.dumps(payload), headers=headers, timeout=5)
         # Log response code
         log_event(
             severity=6,
-            message=f"Webhook sent to {webhook.url}, response code: {resp.status_code}",
+            message=f"Webhook sent to {url}, response code: {resp.status_code}",
             event_type=event or getattr(webhook, 'event_type', None),
             user_id=getattr(webhook, 'created_by', None),
             source_ip=None,
@@ -83,7 +84,7 @@ def trigger_webhook(webhook, attestation_data, event=None):
     except Exception as e:
         log_event(
             severity=3,  # Error
-            message=f"Error sending webhook to {webhook.url}: {e}",
+            message=f"Error sending webhook to {url}: {e}",
             event_type=event or getattr(webhook, 'event_type', None),
             user_id=getattr(webhook, 'created_by', None),
             source_ip=None,
@@ -91,17 +92,116 @@ def trigger_webhook(webhook, attestation_data, event=None):
         )
         return None
 
-import json
-import io
-from flask import Blueprint, request, jsonify, send_file
-from flask_jwt_extended import get_jwt_identity
-from models import Webhook, User, Attestation
-from app import db
-from auth import jwt_required_with_logging
-import requests
-from datetime import datetime
-from logging_utils import log_event
-from flask import current_app
+def render_template_string(template, context=None, attestation=None, webhook=None):
+    """
+    Replace all "--varname--" patterns in the template string with their resolved values.
+    All replacements are done at the string level, then the result is parsed as JSON if possible.
+    """
+    import json
+
+    context = context or {}
+    attestation = attestation or {}
+    labels = context.get('labels', []) if isinstance(context.get('labels', []), list) else []
+    storage_path = attestation.get('storage_path')
+    storage_url = attestation.get('storage_url')
+    attestation_loaded = {}
+
+    def get_label_value(label_name):
+        for label in labels:
+            if isinstance(label, str) and label.startswith(f"{label_name}="):
+                return label.split("=", 1)[1]
+        return ""
+
+    def resolve_var(varname):
+        # Built-in
+        if varname == "secret":
+            return getattr(webhook, "secret", "") if webhook else ""
+        if varname == "product-key":
+            return attestation.get("product_key", "")
+        if varname == "product-version":
+            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:
+                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 ""
+        # Context fields
+        if varname.startswith("labels."):
+            label_name = varname.split(".", 1)[1]
+            return get_label_value(label_name)
+        if varname in context:
+            return str(context[varname])
+        # Not found
+        return ""
+
+    # Always operate on strings
+    if not isinstance(template, str):
+        return template
+
+    pattern = r'--([a-zA-Z0-9_.-]+)--'
+
+    def json_escape(val):
+        # For dict/list, dump as JSON; for bytes, decode; else str
+        if isinstance(val, (dict, list)):
+            return json.dumps(val)
+        if isinstance(val, bytes):
+            try:
+                return val.decode()
+            except Exception:
+                return str(val)
+        return str(val)
+
+    def replacer(match):
+        varname = match.group(1)
+        value = resolve_var(varname)
+        return json_escape(value)
+
+    replaced = re.sub(pattern, replacer, template)
+
+    # Try to parse as JSON, fallback to string
+    try:
+        return json.loads(replaced)
+    except Exception:
+        return replaced
 
 webhooks_bp = Blueprint('webhooks', __name__)
 

templates/webhook_form.html

@@ -81,9 +81,20 @@
                             <label for="content" class="form-label">Content Template (JSON)</label>
                             <textarea class="form-control" id="content" name="content" rows="5" placeholder='{"key":"value", "attestation":{attestation_details}}'>{% if webhook and webhook.content %}{{ webhook.content }}{% else %}{"attestation":{attestation_details}}{% endif %}</textarea>
                             <small class="form-text text-muted">
-                                Use <code>"{attestation_details}"</code> (with quotes) in your JSON template to inject the attestation dictionary.<br>
-                                You can also use <code>{event}</code> for the event type and <code>{webhook_id}</code> for the webhook ID in your template.<br>
-                                Example: <code>{"attestation": "{attestation_details}"}</code> will inject the attestation as a dictionary, not a string.
+                                <b>Templating:</b> You can use variables in the form <code>"--varname--"</code> (double quote, two dashes, name, two dashes, double quote) anywhere in the URL, headers, or content.<br>
+                                <b>Supported variables:</b>
+                                <ul>
+                                    <li><code>secret</code>, <code>product-key</code>, <code>product-version</code>, <code>attestation_details</code></li>
+                                    <li><code>raw_attestation</code>, <code>attestation_json</code>, <code>attestation_predicate</code>, <code>attestation_predicate_b64</code> (loads attestation from storage only if used)</li>
+                                    <li>Any context field: <code>--contextkey--</code> (empty string if missing)</li>
+                                    <li>Labels: <code>--labels.labelname--</code> (gets value for label <code>labelname</code>, empty if not found)</li>
+                                </ul>
+                                <b>Examples:</b>
+                                <ul>
+                                    <li><code>{"Authorization": "Bearer \"--secret--\""}</code> → <code>{"Authorization": "Bearer ghpat-1232455"}</code></li>
+                                    <li><code>{"info": "--attestation_details--"}</code> → <code>{"info": {"id": 5, "context": {...}}}</code></li>
+                                </ul>
+                                If a variable is not found, it is replaced with an empty string.
                             </small>
                         </div>
                         <div class="mb-3">

templates/webhooks.html

@@ -180,9 +180,20 @@
                 <label for="webhookContent" class="form-label">Content Template (JSON)</label>
                 <textarea class="form-control" id="webhookContent" name="content" rows="5" placeholder='{"attestation":{attestation_details}}'></textarea>
                 <small class="form-text text-muted">
-                  Use <code>"{attestation_details}"</code> (with quotes) in your JSON template to inject the attestation dictionary.<br>
-                  You can also use <code>{event}</code> for the event type and <code>{webhook_id}</code> for the webhook ID in your template.<br>
-                  Example: <code>{"attestation": "{attestation_details}"}</code> will inject the attestation as a dictionary, not a string.
+                  <b>Templating:</b> You can use variables in the form <code>"--varname--"</code> (double quote, two dashes, name, two dashes, double quote) anywhere in the URL, headers, or content.<br>
+                  <b>Supported variables:</b>
+                  <ul>
+                      <li><code>secret</code>, <code>product-key</code>, <code>product-version</code>, <code>attestation_details</code></li>
+                      <li><code>raw_attestation</code>, <code>attestation_json</code>, <code>attestation_predicate</code>, <code>attestation_predicate_b64</code> (loads attestation from storage only if used)</li>
+                      <li>Any context field: <code>--contextkey--</code> (empty string if missing)</li>
+                      <li>Labels: <code>--labels.labelname--</code> (gets value for label <code>labelname</code>, empty if not found)</li>
+                  </ul>
+                  <b>Examples:</b>
+                  <ul>
+                      <li><code>{"Authorization": "Bearer \"--secret--\""}</code> → <code>{"Authorization": "Bearer ghpat-1232455"}</code></li>
+                      <li><code>{"info": "--attestation_details--"}</code> → <code>{"info": {"id": 5, "context": {...}}}</code></li>
+                  </ul>
+                  If a variable is not found, it is replaced with an empty string.
                 </small>
               </div>
               <div class="mb-3">