| |
| 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. |
| - Replaces {secret} in headers with webhook.secret |
| - Replaces {attestation_details} in content with attestation_data |
| """ |
| import json, requests |
| |
| headers = {} |
| context = attestation_data.get("context", {}) if isinstance(attestation_data, dict) else {} |
| |
| if webhook.headers: |
| try: |
| user_headers = json.loads(webhook.headers) |
| for k, v in user_headers.items(): |
| 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 [])): |
| headers['X-Webhook-Secret'] = webhook.secret |
| |
| if webhook.content: |
| try: |
| content_str = webhook.content |
| payload_str = render_template_string(content_str, context, attestation_data, webhook) |
| try: |
| payload = json.loads(payload_str) |
| except Exception: |
| payload = payload_str |
| except Exception as e: |
| current_app.logger.error(f"Error parsing webhook content: {e}") |
| return None |
| else: |
| payload = None |
| |
| 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_event( |
| severity=6, |
| 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(url, data=json.dumps(payload) if payload else None, headers=headers, timeout=5) |
| |
| log_event( |
| severity=6, |
| 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, |
| app_name='webhook-sender' |
| ) |
| return resp |
| except Exception as e: |
| log_event( |
| severity=3, |
| 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, |
| app_name='webhook-sender' |
| ) |
| return None |
|
|
| 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): |
| |
| 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 |
|
|
| |
| 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") |
|
|
| |
| 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 "" |
|
|
| |
| 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 "" |
|
|
| |
| if varname == "attestation_raw": |
| return raw_content |
|
|
| |
| predicate, content_type, is_json, error = get_attestation_file_content(att_obj) |
| if error or predicate is None: |
| return "" |
| if varname == "attestation_predicate": |
| 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 "" |
|
|
| |
| if varname.startswith("labels."): |
| label_name = varname.split(".", 1)[1] |
| return get_label_value(label_name) |
| if varname in context: |
| return str(context[varname]) |
| |
| return "" |
|
|
| |
| if not isinstance(template, str): |
| return template |
|
|
| pattern = r'--([a-zA-Z0-9_.-]+)--' |
|
|
| def json_escape(val): |
| |
| 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) |
| |
| if isinstance(replaced, str): |
| return replaced |
| return str(replaced) |
|
|
| webhooks_bp = Blueprint('webhooks', __name__) |
|
|
| @webhooks_bp.route('/webhooks', methods=['POST']) |
| @jwt_required_with_logging |
| def create_webhook(): |
| """Create a new webhook for the current user""" |
| data = request.get_json() |
| if not data: |
| return jsonify({'error': 'JSON data required'}), 400 |
|
|
| name = data.get('name') |
| trigger_type = data.get('trigger_type', 'badge') |
| badge = data.get('badge') |
| trigger_non_zero = bool(data.get('trigger_non_zero', False)) |
| attestation_conditions = data.get('attestation_conditions') |
| url = data.get('url', 'https://httpbin.org/post') |
| secret = data.get('secret') |
| is_active = data.get('is_active', True) |
| headers = data.get('headers', '{}') |
| content = data.get('content') |
| method = data.get('method', 'POST') |
|
|
| if not name: |
| return jsonify({'error': 'name is required'}), 400 |
| if trigger_type == 'badge' and (not badge or not url): |
| return jsonify({'error': 'badge and url are required for badge trigger'}), 400 |
| if trigger_type == 'attestation-condition' and not url: |
| return jsonify({'error': 'url is required for attestation-condition trigger'}), 400 |
|
|
| |
| if not content: |
| content = json.dumps({"attestation": "{attestation_details}"}) |
|
|
| user_id = int(get_jwt_identity()) |
| webhook = Webhook( |
| name=name, |
| trigger_type=trigger_type, |
| badge=badge if trigger_type == 'badge' else None, |
| trigger_non_zero=trigger_non_zero if trigger_type == 'badge' else None, |
| attestation_conditions=attestation_conditions if trigger_type == 'attestation-condition' else None, |
| url=url, |
| secret=secret, |
| created_by=user_id, |
| created_at=datetime.utcnow(), |
| is_active=is_active, |
| headers=headers, |
| content=content, |
| method=method |
| ) |
| from sqlalchemy.exc import IntegrityError |
| db.session.add(webhook) |
| try: |
| db.session.commit() |
| except IntegrityError: |
| db.session.rollback() |
| return jsonify({'error': 'Webhook name must be unique'}), 400 |
| return jsonify({'webhook': webhook.to_dict()}), 201 |
|
|
| @webhooks_bp.route('/webhooks', methods=['GET']) |
| @jwt_required_with_logging |
| def list_webhooks(): |
| """List all webhooks for the current user""" |
| user_id = int(get_jwt_identity()) |
| webhooks = Webhook.query.filter_by(created_by=user_id).all() |
| return jsonify({'webhooks': [w.to_dict() for w in webhooks]}) |
|
|
| @webhooks_bp.route('/webhooks/<int:webhook_id>', methods=['DELETE']) |
| @jwt_required_with_logging |
| def delete_webhook(webhook_id): |
| """Delete a webhook by ID""" |
| user_id = int(get_jwt_identity()) |
| webhook = Webhook.query.filter_by(id=webhook_id, created_by=user_id).first() |
| if not webhook: |
| return jsonify({'error': 'Webhook not found'}), 404 |
| db.session.delete(webhook) |
| db.session.commit() |
| return jsonify({'result': 'deleted'}) |
|
|
| @webhooks_bp.route('/webhooks/test/<int:webhook_id>', methods=['POST']) |
| @jwt_required_with_logging |
| def test_webhook(webhook_id): |
| """Test a webhook by sending a sample payload""" |
| user_id = int(get_jwt_identity()) |
| webhook = Webhook.query.filter_by(id=webhook_id, created_by=user_id).first() |
| if not webhook: |
| return jsonify({'error': 'Webhook not found'}), 404 |
|
|
| |
| attestation = Attestation.query.filter_by(user_id=user_id, deleted=False).order_by(Attestation.created_at.desc()).first() |
| if attestation: |
| attestation_data = attestation.to_dict() |
| attestation_data['download_url'] = attestation.storage_url |
| else: |
| attestation_data = { |
| 'id': 0, |
| 'key': 'sample', |
| 'content_type': webhook.attestation_type or 'cyclone-dx', |
| 'product_key': webhook.product_key or 'sample-product', |
| 'metadata': {}, |
| 'download_url': 'https://example.com/download/sample' |
| } |
|
|
| resp = trigger_webhook(webhook, attestation_data, event=webhook.badge) |
| if resp is not None: |
| return jsonify({ |
| 'status_code': resp.status_code, |
| 'response': resp.text |
| }) |
| else: |
| return jsonify({'error': 'Webhook delivery failed'}), 500 |
| |
| @webhooks_bp.route('/webhooks/<int:webhook_id>', methods=['PUT']) |
| @jwt_required_with_logging |
| def update_webhook(webhook_id): |
| """Update a webhook by ID""" |
| user_id = int(get_jwt_identity()) |
| webhook = Webhook.query.filter_by(id=webhook_id, created_by=user_id).first() |
| if not webhook: |
| return jsonify({'error': 'Webhook not found'}), 404 |
| data = request.get_json() |
| if not data: |
| return jsonify({'error': 'JSON data required'}), 400 |
| webhook.name = data.get('name', webhook.name) |
| webhook.trigger_type = data.get('trigger_type', webhook.trigger_type) |
| webhook.badge = data.get('badge', webhook.badge) if webhook.trigger_type == 'badge' else None |
| webhook.trigger_non_zero = bool(data.get('trigger_non_zero', webhook.trigger_non_zero)) if webhook.trigger_type == 'badge' else None |
| webhook.attestation_conditions = data.get('attestation_conditions', webhook.attestation_conditions) if webhook.trigger_type == 'attestation-condition' else None |
| webhook.url = data.get('url', webhook.url) |
| webhook.secret = data.get('secret', webhook.secret) |
| webhook.is_active = data.get('is_active', webhook.is_active) |
| webhook.headers = data.get('headers', webhook.headers) |
| webhook.content = data.get('content', webhook.content) |
| webhook.method = data.get('method', webhook.method) |
| from sqlalchemy.exc import IntegrityError |
| try: |
| db.session.commit() |
| except IntegrityError: |
| db.session.rollback() |
| return jsonify({'error': 'Webhook name must be unique'}), 400 |
| return jsonify({'webhook': webhook.to_dict()}), 200 |
|
|
| @webhooks_bp.route('/webhooks/<int:webhook_id>/json', methods=['GET']) |
| @jwt_required_with_logging |
| def webhook_json(webhook_id): |
| """Return webhook settings as JSON for the given webhook ID""" |
| user_id = int(get_jwt_identity()) |
| webhook = Webhook.query.filter_by(id=webhook_id, created_by=user_id).first() |
| if not webhook: |
| return jsonify({'error': 'Webhook not found'}), 404 |
| return jsonify(webhook.to_dict()) |
|
|
| @webhooks_bp.route('/webhooks/test-receiver', methods=['POST']) |
| def webhook_test_receiver(): |
| """ |
| Endpoint to receive test webhooks and log the event using log_event. |
| Expects: |
| - JSON body (payload) |
| - 'X-Webhook-Secret' header with a valid token |
| """ |
| |
| secret = request.headers.get('X-Webhook-Secret') |
| if not secret: |
| return jsonify({'error': 'Missing X-Webhook-Secret header'}), 401 |
|
|
| |
| |
| |
| |
| |
|
|
| |
| try: |
| payload = request.get_json(force=True, silent=True) or {} |
| except Exception: |
| payload = {} |
|
|
| log_event( |
| severity=6, |
| message=f"Webhook received: {payload}", |
| event_type=payload.get('event', 'webhook_test'), |
| user_id=None, |
| source_ip=request.remote_addr, |
| app_name='webhook-test-receiver' |
| ) |
| return jsonify({'result': 'Webhook received and logged'}), 200 |
|
|
| |
| @webhooks_bp.route('/webhooks/settings/download', methods=['GET']) |
| @jwt_required_with_logging |
| def download_webhook_settings(): |
| """Download webhooks settings as JSON for the current user.""" |
| user_id = int(get_jwt_identity()) |
| webhooks = Webhook.query.filter_by(created_by=user_id).order_by(Webhook.created_at.desc()).all() |
| data = {wh.name: wh.to_dict() for wh in webhooks} |
| json_bytes = json.dumps(data, indent=2).encode('utf-8') |
| return send_file( |
| io.BytesIO(json_bytes), |
| mimetype='application/json', |
| as_attachment=True, |
| download_name='webhooks_settings.json' |
| ) |
|
|
| @webhooks_bp.route('/webhooks/settings/upload', methods=['POST']) |
| @jwt_required_with_logging |
| def upload_webhook_settings(): |
| """Upload webhook settings from JSON. Updates existing or adds new entries.""" |
| user_id = int(get_jwt_identity()) |
| try: |
| if request.is_json: |
| data = request.get_json() |
| else: |
| file = request.files.get('file') |
| if not file: |
| return jsonify({'error': 'No file uploaded'}), 400 |
| data = json.load(file) |
| except Exception as e: |
| return jsonify({'error': f'Invalid JSON: {e}'}), 400 |
|
|
| required_fields = {'event_type', 'product_key', 'attestation_type', 'url', 'secret', 'is_active', 'headers', 'content', 'method'} |
| accepted, rejected = [], [] |
| for name, cfg in data.items(): |
| if not isinstance(cfg, dict) or not required_fields.issubset(cfg): |
| rejected.append({'name': name, 'reason': 'Missing required fields'}) |
| continue |
| wh = Webhook.query.filter_by(name=name).first() |
| if wh: |
| wh.event_type = cfg['event_type'] |
| wh.product_key = cfg['product_key'] |
| wh.attestation_type = cfg['attestation_type'] |
| wh.url = cfg['url'] |
| wh.secret = cfg['secret'] |
| wh.is_active = cfg['is_active'] |
| wh.headers = cfg['headers'] |
| wh.content = cfg['content'] |
| wh.method = cfg['method'] |
| wh.created_by = user_id |
| wh.created_at = datetime.utcnow() |
| db.session.commit() |
| accepted.append({'name': name, 'action': 'updated'}) |
| else: |
| try: |
| new_wh = Webhook( |
| name=name, |
| event_type=cfg['event_type'], |
| product_key=cfg['product_key'], |
| attestation_type=cfg['attestation_type'], |
| url=cfg['url'], |
| secret=cfg['secret'], |
| created_by=user_id, |
| created_at=datetime.utcnow(), |
| is_active=cfg['is_active'], |
| headers=cfg['headers'], |
| content=cfg['content'], |
| method=cfg['method'] |
| ) |
| db.session.add(new_wh) |
| db.session.commit() |
| accepted.append({'name': name, 'action': 'added'}) |
| except Exception as e: |
| db.session.rollback() |
| rejected.append({'name': name, 'reason': f'Error adding: {e}'}) |
| return jsonify({'accepted': accepted, 'rejected': rejected}) |
|
|
