from job_manager import job_manager
import os
import json
import urllib.request
import re
import io
import csv
import uuid
from flask import Blueprint, request, jsonify, send_file, current_app
from flask_jwt_extended import get_jwt_identity
from werkzeug.utils import secure_filename
from sqlalchemy import func, cast, String, Text, BigInteger, Numeric
from models import Attestation, UploadSession, User, Webhook, PolicyResult, Product
from app import db
from auth import jwt_required_with_logging, optional_jwt_with_logging, get_current_user_safe
from api.badges import is_admin_user
from storage import get_storage_backend
from logging_utils import log_event
import uuid
from datetime import datetime
import requests
evidence_bp = Blueprint('evidence', __name__)
def extract_policy_fields_from_context(context):
"""Extract policy-specific fields from context JSON for deduplication"""
if not context:
return None, None, None
initiative_id = context.get('initiative_id')
gate_type = context.get('gate_type')
gate_name = context.get('gate_name')
return initiative_id, gate_type, gate_name
def find_duplicate_attestation(product_key, product_version, labels, user_id, target_type=None, input_name=None, input_version=None, context=None):
"""Find existing attestation with same product_key, product_version, labels, target_type, input_name, and input_version for current user.
For policy results, also compares initiative_id, gate_type, and gate_name."""
if not product_key or not product_version:
return None
query = Attestation.query.filter_by(
product_key=product_key,
product_version=product_version,
deleted=False,
user_id=user_id
)
# Add optional filters for target_type, input_name, and input_version if provided
if target_type:
query = query.filter_by(target_type=target_type)
if input_name:
query = query.filter_by(target_name=input_name)
if input_version:
# We'll need to filter on context data since there's no direct column for input_version
query = query.filter(Attestation.context['input_version'].astext == input_version)
# Extract policy fields from context
initiative_id, gate_type, gate_name = None, None, None
if context and target_type == "policy-results":
initiative_id, gate_type, gate_name = extract_policy_fields_from_context(context)
# Check labels match and policy fields if applicable
if labels:
# Convert labels to sorted list for comparison
labels_list = sorted(labels) if isinstance(labels, list) else []
for attestation in query.all():
existing_labels = sorted(attestation.labels or [])
# If labels don't match, continue to next attestation
if existing_labels != labels_list:
continue
# For policy results, check additional fields
if target_type == "policy-results" and attestation.context:
att_initiative_id, att_gate_type, att_gate_name = extract_policy_fields_from_context(attestation.context)
# If any of the policy fields don't match, continue to next attestation
if initiative_id != att_initiative_id or gate_type != att_gate_type or gate_name != att_gate_name:
continue
# If we got here, all fields match
return attestation
else:
# Looking for attestations with no labels
for attestation in query.all():
if attestation.labels:
continue
# For policy results, check additional fields
if target_type == "policy-results" and attestation.context:
att_initiative_id, att_gate_type, att_gate_name = extract_policy_fields_from_context(attestation.context)
# If any of the policy fields don't match, continue to next attestation
if initiative_id != att_initiative_id or gate_type != att_gate_type or gate_name != att_gate_name:
continue
# If we got here, all fields match
return attestation
return None
@evidence_bp.route('/evidence', methods=['POST'])
@jwt_required_with_logging
def create_attestation():
"""Create attestation with metadata and get upload URL - follows swagger spec"""
try:
data = request.get_json()
if not data:
return jsonify({'error': 'JSON data required'}), 400
# Extract required fields from swagger spec
key = data.get('key')
content_type = data.get('content_type')
context_type = data.get('context_type')
context = data.get('context', {})
# Log received content type for debugging
log_event(
severity=6,
message=f"Received content_type: '{content_type}' for key: {key}",
event_type="content_type_debug",
user_id=None
)
if not key:
return jsonify({'error': 'key is required'}), 400
# Allow empty content_type - default to application/octet-stream
if not content_type:
content_type = 'application/octet-stream'
if not context_type:
return jsonify({'error': 'context_type is required'}), 400
current_user_id = int(get_jwt_identity())
# Extract context fields for deduplication
product_key = context.get('product_key') or context.get('name')
product_version = context.get('product_version') or context.get('version')
labels = context.get('labels', [])
target_type = context.get('target_type', None)
input_name = context.get('input_name', None)
input_version = context.get('input_version', None)
# Check for duplicate attestation for current user
existing_attestation = find_duplicate_attestation(product_key, product_version, labels, current_user_id, target_type, input_name, input_version, context)
if existing_attestation:
# Mark old attestation as deleted
existing_attestation.deleted = True
existing_attestation.updated_at = datetime.utcnow()
# Mark all policy results of this attestation as deleted
from models import PolicyResult
for policy_result in existing_attestation.policy_results:
policy_result.deleted = True
log_event(
severity=6,
message=f"Replacing existing attestation ID {existing_attestation.id}",
event_type="attestation_replaced",
user_id=current_user_id
)
# Create new attestation in pending state
attestation = Attestation()
attestation.key = key
attestation.target_name = context.get('input_name', 'unknown')
attestation.target_type = context.get('target_type', 'unknown')
attestation.content_type = content_type
attestation.context_type = context_type
attestation.tool = context.get('tool')
attestation.sig_status = 'signed' if context.get('signed', False) else 'unsigned'
attestation.product_key = product_key
attestation.product_version = product_version
attestation.labels = labels
attestation.storage_path = None # Will be set when generating upload URL
attestation.storage_url = None # Will be set when upload is finished
attestation.file_size = None # Will be set when upload is finished
attestation.context = context
attestation.user_id = current_user_id
attestation.team_id = context.get('team_id')
attestation.job_ids = context.get('job_ids', [])
attestation.pipeline_run = context.get('pipeline_run')
attestation.upload_pending = True # Mark as pending upload
# Update relationship between attestations
if existing_attestation:
attestation.replaced_attestation_id = existing_attestation.id
existing_attestation.replaced_by_attestation_id = None # Will be set after attestation gets an ID
deliverable_value = context.get('deliverable', False)
if isinstance(deliverable_value, str):
attestation.deliverable = deliverable_value.lower() == "true"
else:
attestation.deliverable = bool(deliverable_value)
db.session.add(attestation)
db.session.flush() # Get the ID without committing
# Update replaced_by_attestation_id in existing attestation if it exists
if existing_attestation:
existing_attestation.replaced_by_attestation_id = attestation.id
# Generate upload URL using the attestation ID as file_id
storage = get_storage_backend()
file_name = f"{key}" # Use key as filename
# Pass request.url_root to support proxy/k8s scenarios
upload_info = storage.generate_upload_url(file_name, content_type, str(attestation.id), request.url_root)
# Update attestation with storage path
attestation.storage_path = upload_info['storage_path']
# Create upload session for tracking
upload_session = UploadSession()
upload_session.session_id = str(attestation.id)
upload_session.user_id = current_user_id
upload_session.status = 'pending'
upload_session.file_name = file_name
upload_session.upload_url = upload_info['upload_url']
upload_session.storage_path = upload_info['storage_path']
db.session.add(upload_session)
db.session.commit()
log_event(
severity=6,
message=f"Attestation created with upload pending: {attestation.key} for {attestation.target_name}",
event_type="attestation_created_pending",
user_id=current_user_id
)
# Return response matching swagger spec
response = {
'file_id': attestation.id,
'presigned_url': upload_info['upload_url']
}
if existing_attestation:
response['message'] = 'Existing attestation will be replaced'
return jsonify(response)
except Exception as e:
db.session.rollback()
try:
user_id = int(get_jwt_identity())
except:
user_id = None
log_event(
severity=3,
message=f"Failed to create attestation: {str(e)}",
event_type="attestation_error",
user_id=user_id
)
return jsonify({'error': str(e)}), 500
def _compare_scalar(query, key: str, value, Attestation):
"""Generic JSON scalar comparison for both engines."""
if db.engine.dialect.name == 'postgresql':
# Postgres JSON (not jsonb): use ->> for text
expr = Attestation.context.op('->>')(key)
if isinstance(value, bool):
# ->> returns 'true'/'false' strings
return query.filter(expr == ('true' if value else 'false'))
if isinstance(value, int):
return query.filter(cast(expr, BigInteger) == value)
if isinstance(value, float):
return query.filter(cast(expr, Numeric) == value)
# default string compare
return query.filter(expr == str(value))
# SQLite (json1)
# json_extract returns SQL text; cast for numeric when needed
json_path = f'$.{key}'
if isinstance(value, bool):
# SQLite json1 returns 1/0 for booleans via json_extract
return query.filter(func.json_extract(Attestation.context, json_path) == (1 if value else 0))
if isinstance(value, (int, float)):
return query.filter(func.json_extract(Attestation.context, json_path) == value)
return query.filter(func.json_extract(Attestation.context, json_path) == str(value))
@evidence_bp.route('/evidence/list', methods=['POST'])
@jwt_required_with_logging
def list_evidence():
"""Get a list of evidences for a query - follows swagger spec"""
try:
data = request.get_json()
if not data:
data = {}
current_user_id = int(get_jwt_identity())
# Build query - filter by current user and exclude pending uploads
query = Attestation.query.filter_by(
deleted=False,
user_id=current_user_id,
upload_pending=False
)
# Apply filters from AllContext data
for key, value in data.items():
if key == 'name' or key == 'product_key':
query = query.filter(Attestation.product_key == value)
elif key == 'product_version':
query = query.filter(Attestation.product_version == value)
elif key == 'target_name':
query = query.filter(Attestation.target_name == value)
elif key == 'target_type':
query = query.filter(Attestation.target_type == value)
elif key == 'content_type':
query = query.filter(Attestation.content_type == value)
elif key == 'context_type':
query = query.filter(Attestation.context_type == value)
elif key == 'tool':
query = query.filter(Attestation.tool == value)
elif key == 'labels' and value:
# Filter by labels in context
for label in value:
query = query.filter(Attestation.context['labels'].astext.contains(label))
elif value:
query = _compare_scalar(query, key, value, Attestation)
# Get all matching results
attestations = query.all()
# Convert to swagger format
evidences = []
for att in attestations:
evidence = {
'id': att.id,
'key': att.key,
'targetname': att.target_name,
'targettype': att.target_type,
'contenttype': att.content_type,
'contexttype': att.context_type,
'context': att.context,
'tool': att.tool,
'sigstatus': att.sig_status,
'deliverable': att.deliverable,
'userid': att.user_id,
'teamid': att.team_id,
'job_ids': att.job_ids,
'pipelinerun': att.pipeline_run,
'timestamp': {'Time': att.created_at.isoformat(), 'Valid': True} if att.created_at else None,
'deleted': att.deleted
}
evidences.append(evidence)
log_event(
severity=6,
message=f"Listed {len(evidences)} evidences",
event_type="evidences_listed",
user_id=current_user_id
)
# Return response matching swagger spec
return jsonify({
'evidences': evidences,
'all_processed': True
})
except Exception as e:
try:
user_id = int(get_jwt_identity())
except:
user_id = None
log_event(
severity=3,
message=f"Failed to list evidences: {str(e)}",
event_type="list_error",
user_id=user_id
)
return jsonify({'error': str(e)}), 500
@evidence_bp.route('/evidence/<int:file_id>', methods=['GET'])
@jwt_required_with_logging
def get_evidence_details(file_id):
"""Get attestation details and download URL for evidence - extended for frontend modal"""
try:
current_user_id = int(get_jwt_identity())
attestation = Attestation.query.filter_by(
id=file_id,
deleted=False,
user_id=current_user_id,
upload_pending=False
).first()
if not attestation:
return jsonify({'error': 'Evidence not found'}), 404
# Generate fresh download URL
storage = get_storage_backend()
download_url = storage.generate_download_url(attestation.storage_path) if attestation.storage_path else None
log_event(
severity=6,
message=f"Details fetched for evidence {file_id}",
event_type="evidence_details_fetched",
user_id=current_user_id
)
# Return all attestation details plus download URL
data = attestation.to_dict()
data['presigned_url'] = download_url
return jsonify(data)
except Exception as e:
try:
user_id = int(get_jwt_identity())
except:
user_id = None
log_event(
severity=3,
message=f"Failed to get evidence details for {file_id}: {str(e)}",
event_type="evidence_details_error",
user_id=user_id
)
return jsonify({'error': str(e)}), 500
@evidence_bp.route('/evidence/<int:file_id>', methods=['DELETE'])
@jwt_required_with_logging
def delete_evidence(file_id):
"""Delete evidence - follows swagger spec"""
try:
current_user_id = int(get_jwt_identity())
attestation = Attestation.query.filter_by(
id=file_id,
deleted=False,
user_id=current_user_id
).first()
if not attestation:
return jsonify({'error': 'Evidence not found'}), 404
# Delete file from storage
if attestation.storage_path:
storage = get_storage_backend()
storage.delete_file(attestation.storage_path)
# Mark as deleted
attestation.deleted = True
attestation.updated_at = datetime.utcnow()
# Mark all policy results of this attestation as deleted
from models import PolicyResult
for policy_result in attestation.policy_results:
policy_result.deleted = True
db.session.commit()
log_event(
severity=6,
message=f"Evidence deleted: {file_id}",
event_type="evidence_deleted",
user_id=current_user_id
)
# Return 204 No Content as per swagger spec
return '', 204
except Exception as e:
db.session.rollback()
try:
user_id = int(get_jwt_identity())
except:
user_id = None
log_event(
severity=3,
message=f"Failed to delete evidence {file_id}: {str(e)}",
event_type="delete_error",
user_id=user_id
)
return jsonify({'error': str(e)}), 500
@evidence_bp.route('/evidence/upload/<int:file_id>', methods=['PUT'])
def upload_file(file_id):
"""Handle file upload for local storage - uses file_id as authentication token
Note: This endpoint doesn't require JWT auth because it's called with a presigned URL pattern
where the file_id itself serves as the upload token"""
try:
# Get upload session to verify the upload
upload_session = UploadSession.query.filter_by(session_id=str(file_id)).first()
if not upload_session:
return jsonify({'error': 'Upload session not found'}), 404
if upload_session.status != 'pending':
return jsonify({'error': 'Upload session not pending'}), 400
# Get file data from request
file_data = request.get_data()
if not file_data:
return jsonify({'error': 'No file data received'}), 400
# Write file to local storage path
local_path = upload_session.storage_path
storage = get_storage_backend()
# For file storage, get the full local path
if hasattr(storage, 'storage_path'):
full_path = os.path.join(storage.storage_path, local_path)
# SECURITY: Verify the resolved path is within storage directory
real_storage_path = os.path.realpath(storage.storage_path)
real_full_path = os.path.realpath(full_path)
if not real_full_path.startswith(real_storage_path):
log_event(
severity=2,
message=f"Directory traversal attempt detected in upload: {local_path}",
event_type="security_traversal_upload",
user_id=upload_session.user_id
)
return jsonify({'error': 'Invalid file path'}), 400
# Ensure directory exists
os.makedirs(os.path.dirname(full_path), exist_ok=True)
# Write file
with open(full_path, 'wb') as f:
f.write(file_data)
# Update upload session
upload_session.status = 'completed'
upload_session.completed_at = datetime.utcnow()
upload_session.file_size = len(file_data)
db.session.commit()
log_event(
severity=6,
message=f"File uploaded successfully: {file_id}",
event_type="file_uploaded",
user_id=upload_session.user_id
)
return '', 200
else:
return jsonify({'error': 'Local storage not configured'}), 500
except Exception as e:
log_event(
severity=3,
message=f"File upload failed for {file_id}: {str(e)}",
event_type="upload_error"
)
return jsonify({'error': str(e)}), 500
@evidence_bp.route('/evidence/download/<path:storage_path>', methods=['GET'])
def evidence_download(storage_path):
"""Direct file download for any storage type - API access to attestation files
This endpoint is public (no auth required) similar to a pre-signed URL"""
try:
config = current_app.config.get('STORAGE_CONFIG')
storage = get_storage_backend()
# For non-local storage, redirect to presigned URL
if config.STORAGE_TYPE != 'FILE_MOUNT':
download_url = storage.generate_download_url(storage_path)
if not download_url:
return jsonify({'error': 'Failed to generate download URL'}), 500
return jsonify({'download_url': download_url}), 302
# For local storage, serve the file directly
file_path = os.path.join(config.FILE_STORAGE_PATH, storage_path)
if not os.path.exists(file_path):
return jsonify({'error': 'File not found'}), 404
log_event(
severity=6,
message=f"API file downloaded: {storage_path}",
event_type="api_file_downloaded"
)
# Determine content type based on file extension
content_type = None
if file_path.endswith(".json") or file_path.endswith(".json.bin"):
content_type = "application/json"
elif file_path.endswith(".bin"):
# Try to infer from path or default to octet-stream
content_type = "application/octet-stream"
# Serve the file directly
return send_file(
file_path,
mimetype=content_type,
as_attachment=True,
download_name=os.path.basename(file_path)
)
except Exception as e:
log_event(
severity=3,
message=f"API download error: {str(e)}",
event_type="api_download_error"
)
return jsonify({'error': str(e)}), 500
@evidence_bp.route('/evidence/finish', methods=['POST'])
@jwt_required_with_logging
def finish_attestation():
"""Finish attestation upload and finalize the attestation - follows swagger spec"""
try:
data = request.get_json()
if not data:
return jsonify({'error': 'JSON data required'}), 400
file_id = data.get('file_id')
error_message = data.get('error')
if not file_id:
return jsonify({'error': 'file_id required'}), 400
current_user_id = int(get_jwt_identity())
# Find the pending attestation by file_id (which is attestation.id)
attestation = Attestation.query.filter_by(
id=file_id,
user_id=current_user_id,
upload_pending=True,
deleted=False
).first()
if not attestation:
return jsonify({'error': 'Attestation not found or not pending'}), 404
# Get upload session
upload_session = UploadSession.query.filter_by(session_id=str(file_id)).first()
if error_message:
# Mark upload as failed and delete the attestation
if upload_session:
upload_session.status = 'failed'
upload_session.error_message = error_message
upload_session.completed_at = datetime.utcnow()
attestation.deleted = True
attestation.updated_at = datetime.utcnow()
# Mark all policy results of this attestation as deleted (if any)
from models import PolicyResult
for policy_result in attestation.policy_results:
policy_result.deleted = True
log_event(
severity=4,
message=f"Upload failed for file_id {file_id}: {error_message}",
event_type="upload_failed",
user_id=current_user_id
)
db.session.commit()
# Return response matching swagger spec
return jsonify({
'delete_candidates': []
})
# Mark upload as completed and finalize attestation
if upload_session:
upload_session.status = 'completed'
upload_session.completed_at = datetime.utcnow()
# Generate storage URL
storage = get_storage_backend()
storage_url = storage.generate_download_url(attestation.storage_path)
# Finalize attestation
attestation.upload_pending = False
attestation.storage_url = storage_url
attestation.updated_at = datetime.utcnow()
log_event(
severity=6,
message=f"Attestation finalized: {attestation.key} for {attestation.target_name}",
event_type="attestation_completed",
user_id=current_user_id
)
db.session.commit()
# Return response matching swagger spec (return early)
response = jsonify({
'delete_candidates': []
})
# Prepare data for background job
attestation_id = attestation.id
attestation_data = attestation.to_dict()
attestation_data['download_url'] = attestation.storage_url
import threading
from flask import current_app
app_instance = current_app._get_current_object()
t = threading.Thread(target=background_finalize_jobs, args=(attestation_id, attestation_data, attestation.storage_url, current_user_id, app_instance))
t.start()
return response
except Exception as e:
db.session.rollback()
try:
user_id = int(get_jwt_identity())
except:
user_id = None
log_event(
severity=3,
message=f"Failed to finish attestation: {str(e)}",
event_type="finish_attestation_error",
user_id=user_id
)
return jsonify({'error': str(e)}), 500
"""
Ingest SARIF policy result attestation into PolicyResult table.
"""
if not (attestation.content_type and "sarif" in attestation.content_type.lower()):
return
try:
# Import the function here to avoid circular imports
from web.routes import get_attestation_file_content
# Get file content using the unified helper function
content, content_type, is_json, error = get_attestation_file_content(attestation)
if error or not content:
return
# If content was returned as bytes (not JSON), try to parse it
if not is_json and isinstance(content, bytes):
try:
sarif_data = json.loads(content)
except json.JSONDecodeError:
return
else:
# Content is already a JSON object
sarif_data = content
if not sarif_data:
return
# Extract the SARIF content from the predicate if present
# First check standard predicate location for SARIF
sarif_data = sarif_data.get("predicate", {}).get("content", sarif_data)
runs = sarif_data.get("runs", [])
for run in runs:
rules = {}
# SARIF v2: rules are in run['tool']['driver']['rules']
tool = run.get("tool", {})
driver = tool.get("driver", {})
for rule in driver.get("rules", []):
rules[rule.get("id")] = rule
for result in run.get("results", []):
rule_id = result.get("ruleId")
rule = rules.get(rule_id, {})
# Remove 'rule-id:' prefix if it exists. Do this only after fetching the rule !
if rule_id and rule_id.startswith("rule-id:"):
rule_id = rule_id[8:] # Remove 'rule-id:' prefix (8 characters)
rule_result_id = result.get("id", "")
initiative_id = rule_id.split("/")[0] if rule_id and "/" in rule_id else (rule_id or "")
# Extract product_key and product_version from attestation (parsed, not joined)
product_key = getattr(attestation, "product_key", None)
product_version = getattr(attestation, "product_version", None)
policy_result = PolicyResult(
attestation_id=attestation.id,
user_id=attestation.user_id, # Set the user_id from the attestation
attestation_timestamp=attestation.created_at,
initiative_id=initiative_id,
rule_result_id=rule_result_id,
rule_id=rule_id,
rule_name=rule.get("name"),
rule_full_description=(rule.get("fullDescription", {}) or {}).get("text"),
result_kind=result.get("kind"),
result_level=result.get("level"),
result_message=(result.get("message", {}) or {}).get("text"),
rule_json=rule,
result_json=result,
product_key=product_key,
product_version=product_version
)
db.session.add(policy_result)
db.session.commit()
except Exception as e:
db.session.rollback()
return jsonify({'error': str(e)}), 500
@evidence_bp.route('/evidence/policy-results/csv', methods=['GET'])
@jwt_required_with_logging
def policy_results_csv():
"""
API endpoint to download policy results as CSV with filtering options.
Supports the same filtering options as the web UI.
"""
try:
current_user_id = int(get_jwt_identity())
current_user = User.query.get(current_user_id)
is_admin = current_user and is_admin_user(current_user)
# Get filter parameters from query string
rule_id_filter = request.args.get('rule_id', '')
product_filter = request.args.get('product', '')
result_kind_filter = request.args.get('result_kind', '')
result_level_filter = request.args.get('result_level', '')
show_all_users = (request.args.get('show_all_users') == 'true' or request.args.get('show_all_users') == 'on') and is_admin
custom_filename = request.args.get('filename', '')
# Debug information
print(f"API DEBUG - policy_results_csv - request.args: {request.args}")
print(f"API DEBUG - policy_results_csv - show_all_users value: {request.args.get('show_all_users')}")
print(f"API DEBUG - policy_results_csv - show_all_users final: {show_all_users}")
# Build query with filtering
query = PolicyResult.query.filter(PolicyResult.deleted == False)
# Filter by current user unless admin showing all users
if not is_admin or not show_all_users:
query = query.filter(PolicyResult.user_id == current_user_id)
# Apply filters based on query parameters
if rule_id_filter:
query = query.filter(PolicyResult.rule_id.ilike(f"%{rule_id_filter}%"))
if product_filter and product_filter.strip():
query = query.filter(PolicyResult.product_key == product_filter.strip())
if result_kind_filter:
query = query.filter(PolicyResult.result_kind == result_kind_filter)
if result_level_filter:
query = query.filter(PolicyResult.result_level == result_level_filter)
# Order by timestamp, most recent first
results = query.order_by(PolicyResult.attestation_timestamp.desc()).all()
# Generate CSV data
import csv
import io
from werkzeug.utils import secure_filename
# Define CSV headers
csv_headers = [
'Product', 'Version', 'Rule ID', 'Rule Name',
'Result Kind', 'Result Level', 'Message',
'Attestation Timestamp', 'Attestation ID',
'Initiative ID', 'Target Name', 'Target Type'
]
# Add User column for admin when showing all users
if is_admin and show_all_users:
csv_headers.insert(2, 'User')
# Create in-memory file
output = io.StringIO()
writer = csv.writer(output)
# Write headers
writer.writerow(csv_headers)
# Write data rows
for result in results:
rule_id = result.rule_id
if rule_id and rule_id.startswith('rule-id:'):
rule_id = rule_id.replace('rule-id:', '')
row = [
result.product_key or '',
result.product_version or '',
]
# Add username for admin when showing all users
if is_admin and show_all_users:
user = User.query.get(result.user_id) if result.user_id else None
username = user.username if user else '-'
row.append(username)
row.extend([
rule_id,
result.rule_name or '',
result.result_kind or '',
result.result_level or '',
result.result_message or '',
result.attestation_timestamp.isoformat() if result.attestation_timestamp else '',
result.attestation_id or '',
result.initiative_id or '',
'', # target_name not available in PolicyResult
'' # target_type not available in PolicyResult
])
writer.writerow(row)
# Generate filename
if custom_filename and custom_filename.strip():
filename = secure_filename(custom_filename)
# Make sure it has .csv extension
if not filename.lower().endswith('.csv'):
filename += '.csv'
else:
# Generate filename with filters
filename_parts = ['policy_results']
if product_filter:
filename_parts.append(f"product-{product_filter}")
if rule_id_filter:
# Remove any non-alphanumeric characters from the rule ID for the filename
import re
safe_rule_id = re.sub(r'[^\w\-]', '_', rule_id_filter)
filename_parts.append(f"rule-{safe_rule_id}")
if result_kind_filter:
filename_parts.append(f"kind-{result_kind_filter}")
if result_level_filter:
filename_parts.append(f"level-{result_level_filter}")
# Indicate if showing all users
if is_admin and show_all_users:
filename_parts.append("all-users")
# Add timestamp to ensure uniqueness
from datetime import datetime
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
filename_parts.append(timestamp)
filename = f"{'-'.join(filename_parts)}.csv"
# Log the event
log_event(
severity=6,
message=f"Policy results CSV downloaded via API: {filename}",
event_type="policy_results_csv_api_download",
user_id=current_user_id,
source_ip=request.remote_addr
)
# Prepare response
output.seek(0)
response = current_app.response_class(
output.getvalue(),
mimetype='text/csv',
headers={'Content-Disposition': f'attachment; filename="{filename}"'}
)
return response
except Exception as e:
try:
user_id = int(get_jwt_identity())
except:
user_id = None
log_event(
severity=3,
message=f"Failed to generate policy results CSV: {str(e)}",
event_type="policy_results_csv_error",
user_id=user_id
)
return jsonify({'error': str(e)}), 500
@evidence_bp.route('/evidence/team')
@jwt_required_with_logging
def evidence_team():
"""Placeholder endpoint for team evidence"""
return jsonify({"team_id": 1, "team_name": "Teams not supported on on-prem attestation service"})
def ingest_policy_result_from_sarif(attestation):
"""
Ingest SARIF policy result attestation into PolicyResult table.
"""
if not (attestation.content_type and "sarif" in attestation.content_type.lower()):
return
try:
# Import the function here to avoid circular imports
from web.routes import get_attestation_file_content
# Get file content using the unified helper function
content, content_type, is_json, error = get_attestation_file_content(attestation)
if error or not content:
return
# If content was returned as bytes (not JSON), try to parse it
if not is_json and isinstance(content, bytes):
try:
sarif_data = json.loads(content)
except json.JSONDecodeError:
return
else:
# Content is already a JSON object
sarif_data = content
if not sarif_data:
return
# Extract the SARIF content from the predicate if present
# First check standard predicate location for SARIF
sarif_data = sarif_data.get("predicate", {}).get("content", sarif_data)
runs = sarif_data.get("runs", [])
for run in runs:
rules = {}
# SARIF v2: rules are in run['tool']['driver']['rules']
tool = run.get("tool", {})
driver = tool.get("driver", {})
for rule in driver.get("rules", []):
rules[rule.get("id")] = rule
for result in run.get("results", []):
rule_id = result.get("ruleId")
rule = rules.get(rule_id, {})
# Remove 'rule-id:' prefix if it exists. Do this only after fetching the rule !
if rule_id and rule_id.startswith("rule-id:"):
rule_id = rule_id[8:] # Remove 'rule-id:' prefix (8 characters)
rule_result_id = result.get("id", "")
initiative_id = rule_id.split("/")[0] if rule_id and "/" in rule_id else (rule_id or "")
# Extract product_key and product_version from attestation (parsed, not joined)
product_key = getattr(attestation, "product_key", None)
product_version = getattr(attestation, "product_version", None)
from models import PolicyResult
policy_result = PolicyResult(
attestation_id=attestation.id,
user_id=attestation.user_id, # Set the user_id from the attestation
attestation_timestamp=attestation.created_at,
initiative_id=initiative_id,
rule_result_id=rule_result_id,
rule_id=rule_id,
rule_name=rule.get("name"),
rule_full_description=(rule.get("fullDescription", {}) or {}).get("text"),
result_kind=result.get("kind"),
result_level=result.get("level"),
result_message=(result.get("message", {}) or {}).get("text"),
rule_json=rule,
result_json=result,
product_key=product_key,
product_version=product_version
)
from app import db
db.session.add(policy_result)
from app import db
db.session.commit()
except Exception as e:
from app import db
db.session.rollback()
return
# Move background_finalize_jobs to top-level for clarity and to ensure it is defined before use
def background_finalize_jobs(attestation_id, attestation_dict, storage_url, current_user_id, app_instance):
import threading
from models import Attestation, Webhook
from api.webhooks import trigger_webhook
from web.products import update_product_data
from app import db
try:
with app_instance.app_context():
try:
attestation = Attestation.query.get(attestation_id)
# Policy Result SARIF ingestion
try:
if attestation.target_type == "policy-results":
ingest_policy_result_from_sarif(attestation)
except Exception as sarif_exc:
log_event(
severity=3,
message=f"SARIF ingestion failed for attestation {attestation.id}: {str(sarif_exc)}",
event_type="sarif_ingest_error",
user_id=current_user_id
)
# Compute prior badge counts to detect newly created badge entries
prior_product = Product.query.filter_by(
product_key=attestation.product_key,
product_version=attestation.product_version,
deleted=False,
user_id=attestation.user_id
).first()
prior_badges = prior_product.badges if prior_product and isinstance(prior_product.badges, dict) else {}
# Update product badge counts
try:
product = update_product_data(
attestation.product_key,
attestation.product_version,
user_id=attestation.user_id
)
except Exception as upd_exc:
log_event(
severity=3,
message=f"Product update failed: {str(upd_exc)}",
event_type="product_update_error",
user_id=current_user_id
)
product = None
# Trigger webhooks only when a badge count transitions into the hook's trigger condition
if product and isinstance(product.badges, dict):
active_hooks = Webhook.query.filter_by(is_active=True).all()
for hook in active_hooks:
if getattr(hook, "trigger_type", "badge") == "badge":
stats = product.badges.get(hook.badge)
if not stats or 'count' not in stats:
continue
new_count = stats.get('count', 0) or 0
prior_stats = prior_badges.get(hook.badge)
prior_count = (prior_stats.get('count') or 0) if isinstance(prior_stats, dict) else None
if hook.trigger_non_zero:
should_fire = (prior_count is None or prior_count == 0) and new_count > 0
else:
should_fire = (prior_count is None or prior_count > 0) and new_count == 0
if should_fire:
try:
resp = trigger_webhook(hook, attestation_dict, event=hook.badge)
if resp is not None:
log_event(
severity=6,
message=f"Webhook response for '{hook.name}': {resp.status_code} {resp.text}",
event_type='webhook_response',
user_id=current_user_id
)
except Exception as wh_exc:
log_event(
severity=3,
message=f"Webhook trigger failed: {str(wh_exc)}",
event_type="webhook_error",
user_id=current_user_id
)
elif getattr(hook, "trigger_type", "badge") == "attestation-condition":
# Evaluate AND of all conditions on attestation context
context = attestation_dict.get("context", {})
conditions = hook.attestation_conditions or []
all_met = True
for cond in conditions:
field = cond.get("field")
value = cond.get("value")
ctx_val = context.get(field)
if isinstance(ctx_val, list):
if value not in ctx_val:
all_met = False
break
else:
if ctx_val != value:
all_met = False
break
if all_met:
try:
resp = trigger_webhook(hook, attestation_dict, event="attestation-condition")
if resp is not None:
log_event(
severity=6,
message=f"Webhook response for '{hook.name}': {resp.status_code} {resp.text}",
event_type='webhook_response',
user_id=current_user_id
)
except Exception as wh_exc:
log_event(
severity=3,
message=f"Webhook trigger failed: {str(wh_exc)}",
event_type="webhook_error",
user_id=current_user_id
)
finally:
# ensure session is cleaned up after background job
db.session.remove()
except Exception as job_exc:
log_event(
severity=2,
message=f"Background finalize job failed: {str(job_exc)}",
event_type="background_job_error",
user_id=current_user_id
)
from job_manager import job_manager
import os
import json
import urllib.request
import re
import io
import csv
import uuid
from flask import Blueprint, request, jsonify, send_file, current_app
from flask_jwt_extended import get_jwt_identity
from werkzeug.utils import secure_filename
from sqlalchemy import func, cast, String, Text, BigInteger, Numeric
from models import Attestation, UploadSession, User, Webhook, PolicyResult, Product
from app import db
from auth import jwt_required_with_logging, optional_jwt_with_logging, get_current_user_safe
from api.badges import is_admin_user
from storage import get_storage_backend
from logging_utils import log_event
import uuid
from datetime import datetime
import requests
evidence_bp = Blueprint('evidence', __name__)
def extract_policy_fields_from_context(context):
"""Extract policy-specific fields from context JSON for deduplication"""
if not context:
return None, None, None
initiative_id = context.get('initiative_id')
gate_type = context.get('gate_type')
gate_name = context.get('gate_name')
return initiative_id, gate_type, gate_name
def find_duplicate_attestation(product_key, product_version, labels, user_id, target_type=None, input_name=None, input_version=None, context=None):
"""Find existing attestation with same product_key, product_version, labels, target_type, input_name, and input_version for current user.
For policy results, also compares initiative_id, gate_type, and gate_name."""
if not product_key or not product_version:
return None
query = Attestation.query.filter_by(
product_key=product_key,
product_version=product_version,
deleted=False,
user_id=user_id
)
# Add optional filters for target_type, input_name, and input_version if provided
if target_type:
query = query.filter_by(target_type=target_type)
if input_name:
query = query.filter_by(target_name=input_name)
if input_version:
# We'll need to filter on context data since there's no direct column for input_version
query = query.filter(Attestation.context['input_version'].astext == input_version)
# Extract policy fields from context
initiative_id, gate_type, gate_name = None, None, None
if context and target_type == "policy-results":
initiative_id, gate_type, gate_name = extract_policy_fields_from_context(context)
# Check labels match and policy fields if applicable
if labels:
# Convert labels to sorted list for comparison
labels_list = sorted(labels) if isinstance(labels, list) else []
for attestation in query.all():
existing_labels = sorted(attestation.labels or [])
# If labels don't match, continue to next attestation
if existing_labels != labels_list:
continue
# For policy results, check additional fields
if target_type == "policy-results" and attestation.context:
att_initiative_id, att_gate_type, att_gate_name = extract_policy_fields_from_context(attestation.context)
# If any of the policy fields don't match, continue to next attestation
if initiative_id != att_initiative_id or gate_type != att_gate_type or gate_name != att_gate_name:
continue
# If we got here, all fields match
return attestation
else:
# Looking for attestations with no labels
for attestation in query.all():
if attestation.labels:
continue
# For policy results, check additional fields
if target_type == "policy-results" and attestation.context:
att_initiative_id, att_gate_type, att_gate_name = extract_policy_fields_from_context(attestation.context)
# If any of the policy fields don't match, continue to next attestation
if initiative_id != att_initiative_id or gate_type != att_gate_type or gate_name != att_gate_name:
continue
# If we got here, all fields match