678 lines
22 KiB
Python
678 lines
22 KiB
Python
"""
|
|
Webhooks API blueprint.
|
|
|
|
Handles endpoints for managing webhook configurations and viewing delivery logs.
|
|
"""
|
|
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from flask import Blueprint, jsonify, request, current_app
|
|
|
|
from web.auth.decorators import api_auth_required
|
|
from web.models import Webhook, WebhookDeliveryLog, Alert
|
|
from web.services.webhook_service import WebhookService
|
|
from web.services.template_service import get_template_service
|
|
|
|
bp = Blueprint('webhooks_api', __name__)
|
|
|
|
|
|
@bp.route('', methods=['GET'])
|
|
@api_auth_required
|
|
def list_webhooks():
|
|
"""
|
|
List all webhooks with optional filtering.
|
|
|
|
Query params:
|
|
page: Page number (default: 1)
|
|
per_page: Items per page (default: 20)
|
|
enabled: Filter by enabled status (true/false)
|
|
|
|
Returns:
|
|
JSON response with webhooks list
|
|
"""
|
|
# Get query parameters
|
|
page = request.args.get('page', 1, type=int)
|
|
per_page = min(request.args.get('per_page', 20, type=int), 100) # Max 100 items
|
|
enabled = request.args.get('enabled')
|
|
|
|
# Build query
|
|
query = current_app.db_session.query(Webhook)
|
|
|
|
# Apply enabled filter
|
|
if enabled is not None:
|
|
enabled_bool = enabled.lower() == 'true'
|
|
query = query.filter(Webhook.enabled == enabled_bool)
|
|
|
|
# Order by name
|
|
query = query.order_by(Webhook.name)
|
|
|
|
# Paginate
|
|
total = query.count()
|
|
webhooks = query.offset((page - 1) * per_page).limit(per_page).all()
|
|
|
|
# Format response
|
|
webhooks_data = []
|
|
|
|
for webhook in webhooks:
|
|
# Parse JSON fields
|
|
alert_types = json.loads(webhook.alert_types) if webhook.alert_types else None
|
|
severity_filter = json.loads(webhook.severity_filter) if webhook.severity_filter else None
|
|
custom_headers = json.loads(webhook.custom_headers) if webhook.custom_headers else None
|
|
|
|
webhooks_data.append({
|
|
'id': webhook.id,
|
|
'name': webhook.name,
|
|
'url': webhook.url,
|
|
'enabled': webhook.enabled,
|
|
'auth_type': webhook.auth_type,
|
|
'auth_token': '***ENCRYPTED***' if webhook.auth_token else None, # Mask sensitive data
|
|
'custom_headers': custom_headers,
|
|
'alert_types': alert_types,
|
|
'severity_filter': severity_filter,
|
|
'timeout': webhook.timeout,
|
|
'retry_count': webhook.retry_count,
|
|
'created_at': webhook.created_at.isoformat() if webhook.created_at else None,
|
|
'updated_at': webhook.updated_at.isoformat() if webhook.updated_at else None
|
|
})
|
|
|
|
return jsonify({
|
|
'webhooks': webhooks_data,
|
|
'total': total,
|
|
'page': page,
|
|
'per_page': per_page,
|
|
'pages': (total + per_page - 1) // per_page
|
|
})
|
|
|
|
|
|
@bp.route('/<int:webhook_id>', methods=['GET'])
|
|
@api_auth_required
|
|
def get_webhook(webhook_id):
|
|
"""
|
|
Get a specific webhook by ID.
|
|
|
|
Args:
|
|
webhook_id: Webhook ID
|
|
|
|
Returns:
|
|
JSON response with webhook details
|
|
"""
|
|
webhook = current_app.db_session.query(Webhook).filter(Webhook.id == webhook_id).first()
|
|
|
|
if not webhook:
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': f'Webhook {webhook_id} not found'
|
|
}), 404
|
|
|
|
# Parse JSON fields
|
|
alert_types = json.loads(webhook.alert_types) if webhook.alert_types else None
|
|
severity_filter = json.loads(webhook.severity_filter) if webhook.severity_filter else None
|
|
custom_headers = json.loads(webhook.custom_headers) if webhook.custom_headers else None
|
|
|
|
return jsonify({
|
|
'webhook': {
|
|
'id': webhook.id,
|
|
'name': webhook.name,
|
|
'url': webhook.url,
|
|
'enabled': webhook.enabled,
|
|
'auth_type': webhook.auth_type,
|
|
'auth_token': '***ENCRYPTED***' if webhook.auth_token else None,
|
|
'custom_headers': custom_headers,
|
|
'alert_types': alert_types,
|
|
'severity_filter': severity_filter,
|
|
'timeout': webhook.timeout,
|
|
'retry_count': webhook.retry_count,
|
|
'created_at': webhook.created_at.isoformat() if webhook.created_at else None,
|
|
'updated_at': webhook.updated_at.isoformat() if webhook.updated_at else None
|
|
}
|
|
})
|
|
|
|
|
|
@bp.route('', methods=['POST'])
|
|
@api_auth_required
|
|
def create_webhook():
|
|
"""
|
|
Create a new webhook.
|
|
|
|
Request body:
|
|
name: Webhook name (required)
|
|
url: Webhook URL (required)
|
|
enabled: Whether webhook is enabled (default: true)
|
|
auth_type: Authentication type (none, bearer, basic, custom)
|
|
auth_token: Authentication token (encrypted on storage)
|
|
custom_headers: JSON object with custom headers
|
|
alert_types: Array of alert types to filter
|
|
severity_filter: Array of severities to filter
|
|
timeout: Request timeout in seconds (default: 10)
|
|
retry_count: Number of retry attempts (default: 3)
|
|
template: Jinja2 template for custom payload (optional)
|
|
template_format: Template format - 'json' or 'text' (default: json)
|
|
content_type_override: Custom Content-Type header (optional)
|
|
|
|
Returns:
|
|
JSON response with created webhook
|
|
"""
|
|
data = request.get_json() or {}
|
|
|
|
# Validate required fields
|
|
if not data.get('name'):
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': 'name is required'
|
|
}), 400
|
|
|
|
if not data.get('url'):
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': 'url is required'
|
|
}), 400
|
|
|
|
# Validate auth_type
|
|
valid_auth_types = ['none', 'bearer', 'basic', 'custom']
|
|
auth_type = data.get('auth_type', 'none')
|
|
if auth_type not in valid_auth_types:
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': f'Invalid auth_type. Must be one of: {", ".join(valid_auth_types)}'
|
|
}), 400
|
|
|
|
# Validate template_format
|
|
valid_template_formats = ['json', 'text']
|
|
template_format = data.get('template_format', 'json')
|
|
if template_format not in valid_template_formats:
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': f'Invalid template_format. Must be one of: {", ".join(valid_template_formats)}'
|
|
}), 400
|
|
|
|
# Validate template if provided
|
|
template = data.get('template')
|
|
if template:
|
|
template_service = get_template_service()
|
|
is_valid, error_msg = template_service.validate_template(template, template_format)
|
|
if not is_valid:
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': f'Invalid template: {error_msg}'
|
|
}), 400
|
|
|
|
try:
|
|
webhook_service = WebhookService(current_app.db_session)
|
|
|
|
# Encrypt auth_token if provided
|
|
auth_token = None
|
|
if data.get('auth_token'):
|
|
auth_token = webhook_service._encrypt_value(data['auth_token'])
|
|
|
|
# Serialize JSON fields
|
|
alert_types = json.dumps(data['alert_types']) if data.get('alert_types') else None
|
|
severity_filter = json.dumps(data['severity_filter']) if data.get('severity_filter') else None
|
|
custom_headers = json.dumps(data['custom_headers']) if data.get('custom_headers') else None
|
|
|
|
# Create webhook
|
|
webhook = Webhook(
|
|
name=data['name'],
|
|
url=data['url'],
|
|
enabled=data.get('enabled', True),
|
|
auth_type=auth_type,
|
|
auth_token=auth_token,
|
|
custom_headers=custom_headers,
|
|
alert_types=alert_types,
|
|
severity_filter=severity_filter,
|
|
timeout=data.get('timeout', 10),
|
|
retry_count=data.get('retry_count', 3),
|
|
template=template,
|
|
template_format=template_format,
|
|
content_type_override=data.get('content_type_override'),
|
|
created_at=datetime.now(timezone.utc),
|
|
updated_at=datetime.now(timezone.utc)
|
|
)
|
|
|
|
current_app.db_session.add(webhook)
|
|
current_app.db_session.commit()
|
|
|
|
# Parse for response
|
|
alert_types_parsed = json.loads(alert_types) if alert_types else None
|
|
severity_filter_parsed = json.loads(severity_filter) if severity_filter else None
|
|
custom_headers_parsed = json.loads(custom_headers) if custom_headers else None
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': 'Webhook created successfully',
|
|
'webhook': {
|
|
'id': webhook.id,
|
|
'name': webhook.name,
|
|
'url': webhook.url,
|
|
'enabled': webhook.enabled,
|
|
'auth_type': webhook.auth_type,
|
|
'alert_types': alert_types_parsed,
|
|
'severity_filter': severity_filter_parsed,
|
|
'custom_headers': custom_headers_parsed,
|
|
'timeout': webhook.timeout,
|
|
'retry_count': webhook.retry_count,
|
|
'template': webhook.template,
|
|
'template_format': webhook.template_format,
|
|
'content_type_override': webhook.content_type_override,
|
|
'created_at': webhook.created_at.isoformat()
|
|
}
|
|
}), 201
|
|
|
|
except Exception as e:
|
|
current_app.db_session.rollback()
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': f'Failed to create webhook: {str(e)}'
|
|
}), 500
|
|
|
|
|
|
@bp.route('/<int:webhook_id>', methods=['PUT'])
|
|
@api_auth_required
|
|
def update_webhook(webhook_id):
|
|
"""
|
|
Update an existing webhook.
|
|
|
|
Args:
|
|
webhook_id: Webhook ID
|
|
|
|
Request body (all optional):
|
|
name: Webhook name
|
|
url: Webhook URL
|
|
enabled: Whether webhook is enabled
|
|
auth_type: Authentication type
|
|
auth_token: Authentication token
|
|
custom_headers: JSON object with custom headers
|
|
alert_types: Array of alert types
|
|
severity_filter: Array of severities
|
|
timeout: Request timeout
|
|
retry_count: Retry attempts
|
|
template: Jinja2 template for custom payload
|
|
template_format: Template format - 'json' or 'text'
|
|
content_type_override: Custom Content-Type header
|
|
|
|
Returns:
|
|
JSON response with update status
|
|
"""
|
|
webhook = current_app.db_session.query(Webhook).filter(Webhook.id == webhook_id).first()
|
|
|
|
if not webhook:
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': f'Webhook {webhook_id} not found'
|
|
}), 404
|
|
|
|
data = request.get_json() or {}
|
|
|
|
# Validate auth_type if provided
|
|
if 'auth_type' in data:
|
|
valid_auth_types = ['none', 'bearer', 'basic', 'custom']
|
|
if data['auth_type'] not in valid_auth_types:
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': f'Invalid auth_type. Must be one of: {", ".join(valid_auth_types)}'
|
|
}), 400
|
|
|
|
# Validate template_format if provided
|
|
if 'template_format' in data:
|
|
valid_template_formats = ['json', 'text']
|
|
if data['template_format'] not in valid_template_formats:
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': f'Invalid template_format. Must be one of: {", ".join(valid_template_formats)}'
|
|
}), 400
|
|
|
|
# Validate template if provided
|
|
if 'template' in data and data['template']:
|
|
template_format = data.get('template_format', webhook.template_format or 'json')
|
|
template_service = get_template_service()
|
|
is_valid, error_msg = template_service.validate_template(data['template'], template_format)
|
|
if not is_valid:
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': f'Invalid template: {error_msg}'
|
|
}), 400
|
|
|
|
try:
|
|
webhook_service = WebhookService(current_app.db_session)
|
|
|
|
# Update fields if provided
|
|
if 'name' in data:
|
|
webhook.name = data['name']
|
|
if 'url' in data:
|
|
webhook.url = data['url']
|
|
if 'enabled' in data:
|
|
webhook.enabled = data['enabled']
|
|
if 'auth_type' in data:
|
|
webhook.auth_type = data['auth_type']
|
|
if 'auth_token' in data:
|
|
# Encrypt new token
|
|
webhook.auth_token = webhook_service._encrypt_value(data['auth_token'])
|
|
if 'custom_headers' in data:
|
|
webhook.custom_headers = json.dumps(data['custom_headers']) if data['custom_headers'] else None
|
|
if 'alert_types' in data:
|
|
webhook.alert_types = json.dumps(data['alert_types']) if data['alert_types'] else None
|
|
if 'severity_filter' in data:
|
|
webhook.severity_filter = json.dumps(data['severity_filter']) if data['severity_filter'] else None
|
|
if 'timeout' in data:
|
|
webhook.timeout = data['timeout']
|
|
if 'retry_count' in data:
|
|
webhook.retry_count = data['retry_count']
|
|
if 'template' in data:
|
|
webhook.template = data['template']
|
|
if 'template_format' in data:
|
|
webhook.template_format = data['template_format']
|
|
if 'content_type_override' in data:
|
|
webhook.content_type_override = data['content_type_override']
|
|
|
|
webhook.updated_at = datetime.now(timezone.utc)
|
|
current_app.db_session.commit()
|
|
|
|
# Parse for response
|
|
alert_types = json.loads(webhook.alert_types) if webhook.alert_types else None
|
|
severity_filter = json.loads(webhook.severity_filter) if webhook.severity_filter else None
|
|
custom_headers = json.loads(webhook.custom_headers) if webhook.custom_headers else None
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': 'Webhook updated successfully',
|
|
'webhook': {
|
|
'id': webhook.id,
|
|
'name': webhook.name,
|
|
'url': webhook.url,
|
|
'enabled': webhook.enabled,
|
|
'auth_type': webhook.auth_type,
|
|
'alert_types': alert_types,
|
|
'severity_filter': severity_filter,
|
|
'custom_headers': custom_headers,
|
|
'timeout': webhook.timeout,
|
|
'retry_count': webhook.retry_count,
|
|
'template': webhook.template,
|
|
'template_format': webhook.template_format,
|
|
'content_type_override': webhook.content_type_override,
|
|
'updated_at': webhook.updated_at.isoformat()
|
|
}
|
|
})
|
|
|
|
except Exception as e:
|
|
current_app.db_session.rollback()
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': f'Failed to update webhook: {str(e)}'
|
|
}), 500
|
|
|
|
|
|
@bp.route('/<int:webhook_id>', methods=['DELETE'])
|
|
@api_auth_required
|
|
def delete_webhook(webhook_id):
|
|
"""
|
|
Delete a webhook.
|
|
|
|
Args:
|
|
webhook_id: Webhook ID
|
|
|
|
Returns:
|
|
JSON response with deletion status
|
|
"""
|
|
webhook = current_app.db_session.query(Webhook).filter(Webhook.id == webhook_id).first()
|
|
|
|
if not webhook:
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': f'Webhook {webhook_id} not found'
|
|
}), 404
|
|
|
|
try:
|
|
# Delete webhook (delivery logs will be cascade deleted)
|
|
current_app.db_session.delete(webhook)
|
|
current_app.db_session.commit()
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': f'Webhook {webhook_id} deleted successfully'
|
|
})
|
|
|
|
except Exception as e:
|
|
current_app.db_session.rollback()
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': f'Failed to delete webhook: {str(e)}'
|
|
}), 500
|
|
|
|
|
|
@bp.route('/<int:webhook_id>/test', methods=['POST'])
|
|
@api_auth_required
|
|
def test_webhook(webhook_id):
|
|
"""
|
|
Send a test payload to a webhook.
|
|
|
|
Args:
|
|
webhook_id: Webhook ID
|
|
|
|
Returns:
|
|
JSON response with test result
|
|
"""
|
|
webhook = current_app.db_session.query(Webhook).filter(Webhook.id == webhook_id).first()
|
|
|
|
if not webhook:
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': f'Webhook {webhook_id} not found'
|
|
}), 404
|
|
|
|
# Test webhook delivery
|
|
webhook_service = WebhookService(current_app.db_session)
|
|
result = webhook_service.test_webhook(webhook_id)
|
|
|
|
return jsonify({
|
|
'status': 'success' if result['success'] else 'error',
|
|
'message': result['message'],
|
|
'status_code': result['status_code'],
|
|
'response_body': result.get('response_body')
|
|
})
|
|
|
|
|
|
@bp.route('/<int:webhook_id>/logs', methods=['GET'])
|
|
@api_auth_required
|
|
def get_webhook_logs(webhook_id):
|
|
"""
|
|
Get delivery logs for a specific webhook.
|
|
|
|
Args:
|
|
webhook_id: Webhook ID
|
|
|
|
Query params:
|
|
page: Page number (default: 1)
|
|
per_page: Items per page (default: 20)
|
|
status: Filter by status (success/failed)
|
|
|
|
Returns:
|
|
JSON response with delivery logs
|
|
"""
|
|
webhook = current_app.db_session.query(Webhook).filter(Webhook.id == webhook_id).first()
|
|
|
|
if not webhook:
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': f'Webhook {webhook_id} not found'
|
|
}), 404
|
|
|
|
# Get query parameters
|
|
page = request.args.get('page', 1, type=int)
|
|
per_page = min(request.args.get('per_page', 20, type=int), 100)
|
|
status_filter = request.args.get('status')
|
|
|
|
# Build query
|
|
query = current_app.db_session.query(WebhookDeliveryLog).filter(
|
|
WebhookDeliveryLog.webhook_id == webhook_id
|
|
)
|
|
|
|
# Apply status filter
|
|
if status_filter:
|
|
query = query.filter(WebhookDeliveryLog.status == status_filter)
|
|
|
|
# Order by most recent first
|
|
query = query.order_by(WebhookDeliveryLog.delivered_at.desc())
|
|
|
|
# Paginate
|
|
total = query.count()
|
|
logs = query.offset((page - 1) * per_page).limit(per_page).all()
|
|
|
|
# Format response
|
|
logs_data = []
|
|
for log in logs:
|
|
# Get alert info
|
|
alert = current_app.db_session.query(Alert).filter(Alert.id == log.alert_id).first()
|
|
|
|
logs_data.append({
|
|
'id': log.id,
|
|
'alert_id': log.alert_id,
|
|
'alert_type': alert.alert_type if alert else None,
|
|
'alert_message': alert.message if alert else None,
|
|
'status': log.status,
|
|
'response_code': log.response_code,
|
|
'response_body': log.response_body,
|
|
'error_message': log.error_message,
|
|
'attempt_number': log.attempt_number,
|
|
'delivered_at': log.delivered_at.isoformat() if log.delivered_at else None
|
|
})
|
|
|
|
return jsonify({
|
|
'webhook_id': webhook_id,
|
|
'webhook_name': webhook.name,
|
|
'logs': logs_data,
|
|
'total': total,
|
|
'page': page,
|
|
'per_page': per_page,
|
|
'pages': (total + per_page - 1) // per_page
|
|
})
|
|
|
|
|
|
@bp.route('/preview-template', methods=['POST'])
|
|
@api_auth_required
|
|
def preview_template():
|
|
"""
|
|
Preview a webhook template with sample data.
|
|
|
|
Request body:
|
|
template: Jinja2 template string (required)
|
|
template_format: Template format - 'json' or 'text' (default: json)
|
|
|
|
Returns:
|
|
JSON response with rendered template preview
|
|
"""
|
|
data = request.get_json() or {}
|
|
|
|
if not data.get('template'):
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': 'template is required'
|
|
}), 400
|
|
|
|
template = data['template']
|
|
template_format = data.get('template_format', 'json')
|
|
|
|
# Validate template format
|
|
if template_format not in ['json', 'text']:
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': 'Invalid template_format. Must be json or text'
|
|
}), 400
|
|
|
|
try:
|
|
template_service = get_template_service()
|
|
|
|
# Validate template
|
|
is_valid, error_msg = template_service.validate_template(template, template_format)
|
|
if not is_valid:
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': f'Template validation error: {error_msg}'
|
|
}), 400
|
|
|
|
# Render with sample data
|
|
rendered, error = template_service.render_test_payload(template, template_format)
|
|
|
|
if error:
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': f'Template rendering error: {error}'
|
|
}), 400
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'rendered': rendered,
|
|
'format': template_format
|
|
})
|
|
|
|
except Exception as e:
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': f'Failed to preview template: {str(e)}'
|
|
}), 500
|
|
|
|
|
|
@bp.route('/template-presets', methods=['GET'])
|
|
@api_auth_required
|
|
def get_template_presets():
|
|
"""
|
|
Get list of available webhook template presets.
|
|
|
|
Returns:
|
|
JSON response with template presets
|
|
"""
|
|
import os
|
|
|
|
try:
|
|
# Load presets manifest
|
|
presets_file = os.path.join(
|
|
os.path.dirname(__file__),
|
|
'../templates/webhook_presets/presets.json'
|
|
)
|
|
|
|
with open(presets_file, 'r') as f:
|
|
presets_manifest = json.load(f)
|
|
|
|
# Load template contents for each preset
|
|
presets_dir = os.path.join(
|
|
os.path.dirname(__file__),
|
|
'../templates/webhook_presets'
|
|
)
|
|
|
|
for preset in presets_manifest:
|
|
template_file = os.path.join(presets_dir, preset['file'])
|
|
with open(template_file, 'r') as f:
|
|
preset['template'] = f.read()
|
|
# Remove file reference from response
|
|
del preset['file']
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'presets': presets_manifest
|
|
})
|
|
|
|
except FileNotFoundError as e:
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': f'Template presets not found: {str(e)}'
|
|
}), 500
|
|
except Exception as e:
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': f'Failed to load template presets: {str(e)}'
|
|
}), 500
|
|
|
|
|
|
# Health check endpoint
|
|
@bp.route('/health', methods=['GET'])
|
|
def health_check():
|
|
"""
|
|
Health check endpoint for monitoring.
|
|
|
|
Returns:
|
|
JSON response with API health status
|
|
"""
|
|
return jsonify({
|
|
'status': 'healthy',
|
|
'api': 'webhooks',
|
|
'version': '1.0.0-phase5'
|
|
})
|