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 = {'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(): 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 # 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 with templating 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 # 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 {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 response code 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, # Error 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): # 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 # 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 == "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 "" # 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) # Always return a string 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 # Default content template 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/', 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/', 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 # Use a sample attestation or dummy data 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/', 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//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 """ # Get secret from header secret = request.headers.get('X-Webhook-Secret') if not secret: return jsonify({'error': 'Missing X-Webhook-Secret header'}), 401 # Optional: Validate the secret using your token model # from models import Token # token = Token.query.filter_by(token=secret).first() # if not token: # return jsonify({'error': 'Invalid token'}), 403 # Log the event try: payload = request.get_json(force=True, silent=True) or {} except Exception: payload = {} log_event( severity=6, # Informational message=f"Webhook received: {payload}", event_type=payload.get('event', 'webhook_test'), user_id=None, # Optionally extract from token if needed source_ip=request.remote_addr, app_name='webhook-test-receiver' ) return jsonify({'result': 'Webhook received and logged'}), 200 # --- Webhook settings import/export endpoints --- @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})