debian-mirror-gitlab/lib/gitlab/ci/parsers/security/common.rb
2022-05-07 20:08:51 +05:30

304 lines
11 KiB
Ruby

# frozen_string_literal: true
module Gitlab
module Ci
module Parsers
module Security
class Common
SecurityReportParserError = Class.new(Gitlab::Ci::Parsers::ParserError)
def self.parse!(json_data, report, vulnerability_finding_signatures_enabled = false, validate: false)
new(json_data, report, vulnerability_finding_signatures_enabled, validate: validate).parse!
end
def initialize(json_data, report, vulnerability_finding_signatures_enabled = false, validate: false)
@json_data = json_data
@report = report
@validate = validate
@vulnerability_finding_signatures_enabled = vulnerability_finding_signatures_enabled
end
def parse!
set_report_version
return report_data unless valid?
raise SecurityReportParserError, "Invalid report format" unless report_data.is_a?(Hash)
create_scanner
create_scan
create_analyzer
create_findings
report_data
rescue JSON::ParserError
raise SecurityReportParserError, 'JSON parsing failed'
rescue StandardError
raise SecurityReportParserError, "#{report.type} security report parsing failed"
end
private
attr_reader :json_data, :report, :validate
def valid?
if Feature.enabled?(:show_report_validation_warnings, default_enabled: :yaml)
# We want validation to happen regardless of VALIDATE_SCHEMA CI variable
schema_validation_passed = schema_validator.valid?
if validate
schema_validator.errors.each { |error| report.add_error('Schema', error) } unless schema_validation_passed
schema_validation_passed
else
# We treat all schema validation errors as warnings
schema_validator.errors.each { |error| report.add_warning('Schema', error) }
true
end
else
return true if !validate || schema_validator.valid?
schema_validator.errors.each { |error| report.add_error('Schema', error) }
false
end
end
def schema_validator
@schema_validator ||= ::Gitlab::Ci::Parsers::Security::Validators::SchemaValidator.new(report.type, report_data, report.version)
end
def report_data
@report_data ||= Gitlab::Json.parse!(json_data)
end
def report_version
@report_version ||= report_data['version']
end
def top_level_scanner
@top_level_scanner ||= report_data.dig('scan', 'scanner')
end
def scan_data
@scan_data ||= report_data.dig('scan')
end
def analyzer_data
@analyzer_data ||= report_data.dig('scan', 'analyzer')
end
def tracking_data(data)
data['tracking']
end
def create_findings
if report_data["vulnerabilities"]
report_data["vulnerabilities"].each { |finding| create_finding(finding) }
end
end
def create_finding(data, remediations = [])
identifiers = create_identifiers(data['identifiers'])
flags = create_flags(data['flags'])
links = create_links(data['links'])
location = create_location(data['location'] || {})
evidence = create_evidence(data['evidence'])
signatures = create_signatures(tracking_data(data))
if @vulnerability_finding_signatures_enabled && !signatures.empty?
# NOT the signature_sha - the compare key is hashed
# to create the project_fingerprint
highest_priority_signature = signatures.max_by(&:priority)
uuid = calculate_uuid_v5(identifiers.first, highest_priority_signature.signature_hex)
else
uuid = calculate_uuid_v5(identifiers.first, location&.fingerprint)
end
report.add_finding(
::Gitlab::Ci::Reports::Security::Finding.new(
uuid: uuid,
report_type: report.type,
name: finding_name(data, identifiers, location),
compare_key: data['cve'] || '',
location: location,
evidence: evidence,
severity: parse_severity_level(data['severity']),
confidence: parse_confidence_level(data['confidence']),
scanner: create_scanner(data['scanner']),
scan: report&.scan,
identifiers: identifiers,
flags: flags,
links: links,
remediations: remediations,
original_data: data,
metadata_version: report_version,
details: data['details'] || {},
signatures: signatures,
project_id: report.project_id,
vulnerability_finding_signatures_enabled: @vulnerability_finding_signatures_enabled))
end
def create_signatures(tracking)
tracking ||= { 'items' => [] }
signature_algorithms = Hash.new { |hash, key| hash[key] = [] }
tracking['items'].each do |item|
next unless item.key?('signatures')
item['signatures'].each do |signature|
alg = signature['algorithm']
signature_algorithms[alg] << signature['value']
end
end
signature_algorithms.map do |algorithm, values|
value = values.join('|')
signature = ::Gitlab::Ci::Reports::Security::FindingSignature.new(
algorithm_type: algorithm,
signature_value: value
)
if signature.valid?
signature
else
e = SecurityReportParserError.new("Vulnerability tracking signature is not valid: #{signature}")
Gitlab::ErrorTracking.track_exception(e)
nil
end
end.compact
end
def create_scan
return unless scan_data.is_a?(Hash)
report.scan = ::Gitlab::Ci::Reports::Security::Scan.new(scan_data)
end
def set_report_version
report.version = report_version
end
def create_analyzer
return unless analyzer_data.is_a?(Hash)
params = {
id: analyzer_data.dig('id'),
name: analyzer_data.dig('name'),
version: analyzer_data.dig('version'),
vendor: analyzer_data.dig('vendor', 'name')
}
return unless params.values.all?
report.analyzer = ::Gitlab::Ci::Reports::Security::Analyzer.new(**params)
end
def create_scanner(scanner_data = top_level_scanner)
return unless scanner_data.is_a?(Hash)
report.add_scanner(
::Gitlab::Ci::Reports::Security::Scanner.new(
external_id: scanner_data['id'],
name: scanner_data['name'],
vendor: scanner_data.dig('vendor', 'name'),
version: scanner_data.dig('version')))
end
def create_identifiers(identifiers)
return [] unless identifiers.is_a?(Array)
identifiers.map { |identifier| create_identifier(identifier) }.compact
end
def create_identifier(identifier)
return unless identifier.is_a?(Hash)
report.add_identifier(
::Gitlab::Ci::Reports::Security::Identifier.new(
external_type: identifier['type'],
external_id: identifier['value'],
name: identifier['name'],
url: identifier['url']))
end
def create_flags(flags)
return [] unless flags.is_a?(Array)
flags.map { |flag| create_flag(flag) }.compact
end
def create_flag(flag)
return unless flag.is_a?(Hash)
::Gitlab::Ci::Reports::Security::Flag.new(type: flag['type'], origin: flag['origin'], description: flag['description'])
end
def create_links(links)
return [] unless links.is_a?(Array)
links.map { |link| create_link(link) }.compact
end
def create_link(link)
return unless link.is_a?(Hash)
::Gitlab::Ci::Reports::Security::Link.new(name: link['name'], url: link['url'])
end
def parse_severity_level(input)
input&.downcase.then { |value| ::Enums::Vulnerability.severity_levels.key?(value) ? value : 'unknown' }
end
def parse_confidence_level(input)
input&.downcase.then { |value| ::Enums::Vulnerability.confidence_levels.key?(value) ? value : 'unknown' }
end
def create_location(location_data)
raise NotImplementedError
end
def create_evidence(evidence_data)
return unless evidence_data.is_a?(Hash)
::Gitlab::Ci::Reports::Security::Evidence.new(data: evidence_data)
end
def finding_name(data, identifiers, location)
return data['message'] if data['message'].present?
return data['name'] if data['name'].present?
identifier = identifiers.find(&:cve?) || identifiers.find(&:cwe?) || identifiers.first
"#{identifier.name} in #{location&.fingerprint_path}"
end
def calculate_uuid_v5(primary_identifier, location_fingerprint)
uuid_v5_name_components = {
report_type: report.type,
primary_identifier_fingerprint: primary_identifier&.fingerprint,
location_fingerprint: location_fingerprint,
project_id: report.project_id
}
if uuid_v5_name_components.values.any?(&:nil?)
Gitlab::AppLogger.warn(message: "One or more UUID name components are nil", components: uuid_v5_name_components)
return
end
::Security::VulnerabilityUUID.generate(
report_type: uuid_v5_name_components[:report_type],
primary_identifier_fingerprint: uuid_v5_name_components[:primary_identifier_fingerprint],
location_fingerprint: uuid_v5_name_components[:location_fingerprint],
project_id: uuid_v5_name_components[:project_id]
)
end
end
end
end
end
end
Gitlab::Ci::Parsers::Security::Common.prepend_mod_with("Gitlab::Ci::Parsers::Security::Common")