Blame for api/webhooks.py

 
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/<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

    # 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/<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
    """
    # 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})