software vuln
This commit is contained in:
Ruslan Piatrovich
@@ -0,0 +1,202 @@
|
||||
import requests
|
||||
import sys
|
||||
import time
|
||||
import os
|
||||
import psycopg2
|
||||
from psycopg2.extras import execute_values
|
||||
|
||||
JENKINS_URL = "https://jenkins.lmru.tech"
|
||||
AUTH_USER = "lm-sa-devsecops"
|
||||
AUTH_TOKEN = "11b0f9ff6c986b18f0c7fa6e8ae769b66c"
|
||||
|
||||
NVD_API_URL = "https://services.nvd.nist.gov/rest/json/cves/2.0"
|
||||
NVD_API_KEY = "aa054ea4-4a99-4f16-b5ed-1f77b5193d80"
|
||||
|
||||
DB_HOST = "rc1d-f1mdariunacszmi6.mdb.yandexcloud.net"
|
||||
DB_PORT = 6432
|
||||
DB_NAME = "securitydebt"
|
||||
DB_USER = "cicd-monitoring"
|
||||
DB_PASSWORD = "prelude-uT9/HER"
|
||||
|
||||
def get_jenkins_version():
|
||||
"""Получить версию Jenkins."""
|
||||
resp = requests.get(f"{JENKINS_URL}/api/json", auth=(AUTH_USER, AUTH_TOKEN), verify=False)
|
||||
resp.raise_for_status()
|
||||
jenkins_version = resp.headers.get("X-Jenkins")
|
||||
if not jenkins_version:
|
||||
# Если нет в заголовке, пытаемся достать из JSON
|
||||
data = resp.json()
|
||||
jenkins_version = data.get("jenkinsVersion")
|
||||
return jenkins_version
|
||||
|
||||
def get_jenkins_plugins():
|
||||
"""Получить список (shortName, version) для установленных плагинов Jenkins."""
|
||||
url = f"{JENKINS_URL}/pluginManager/api/json?depth=1"
|
||||
resp = requests.get(url, auth=(AUTH_USER, AUTH_TOKEN), verify=False)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
plugins = data.get("plugins", [])
|
||||
plugin_list = []
|
||||
for p in plugins:
|
||||
shortName = p.get("shortName")
|
||||
version = p.get("version")
|
||||
if shortName and version:
|
||||
plugin_list.append((shortName, version))
|
||||
return plugin_list
|
||||
|
||||
def fetch_cves_by_cpe(cpe_name, max_retries=3, delay=5):
|
||||
"""
|
||||
Запросить уязвимости из NVD по cpeName.
|
||||
При ошибках 403/503 делает несколько попыток.
|
||||
"""
|
||||
params = {
|
||||
"cpeName": cpe_name,
|
||||
"resultsPerPage": 50
|
||||
}
|
||||
headers = {
|
||||
"apiKey": NVD_API_KEY
|
||||
}
|
||||
|
||||
for attempt in range(max_retries):
|
||||
resp = requests.get(NVD_API_URL, params=params, headers=headers)
|
||||
|
||||
if resp.status_code == 403:
|
||||
if attempt < max_retries - 1:
|
||||
print(f"Got 403 Forbidden with API key. Retrying in {delay} seconds...")
|
||||
time.sleep(delay)
|
||||
continue
|
||||
else:
|
||||
print("Got 403 Forbidden after maximum retries. Exiting.")
|
||||
sys.exit(1)
|
||||
elif resp.status_code == 503:
|
||||
if attempt < max_retries - 1:
|
||||
print(f"Got 503 Service Unavailable. Retrying in {delay} seconds...")
|
||||
time.sleep(delay)
|
||||
continue
|
||||
else:
|
||||
print("Got 503 Service Unavailable after maximum retries. Exiting.")
|
||||
sys.exit(1)
|
||||
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
return data.get("vulnerabilities", [])
|
||||
|
||||
print("All attempts failed unexpectedly. Exiting.")
|
||||
sys.exit(1)
|
||||
|
||||
def insert_vulnerabilities(conn, name, version, vulnerabilities):
|
||||
"""
|
||||
Вставить или обновить информацию об уязвимостях в PostgreSQL.
|
||||
|
||||
Если нет уязвимостей:
|
||||
- вставляем/обновляем запись с cve_id='NO_VULN'.
|
||||
|
||||
Если уязвимости появились:
|
||||
- удаляем запись 'NO_VULN' (если была),
|
||||
- вставляем все уязвимости с UPSERT.
|
||||
|
||||
Используем ON CONFLICT DO UPDATE для актуализации данных.
|
||||
"""
|
||||
if vulnerabilities:
|
||||
# Удаляем запись NO_VULN, если есть
|
||||
delete_query = """
|
||||
DELETE FROM software_vulnerabilities
|
||||
WHERE name = %s AND version = %s AND cve_id = 'NO_VULN';
|
||||
"""
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(delete_query, (name, version))
|
||||
conn.commit()
|
||||
|
||||
# Вставляем реальные уязвимости
|
||||
records = []
|
||||
for v in vulnerabilities:
|
||||
cve = v.get("cve", {})
|
||||
cve_id = cve.get("id")
|
||||
descs = cve.get("descriptions", [])
|
||||
desc_en = next((d.get("value") for d in descs if d.get("lang") == "en"), "")
|
||||
|
||||
severity = None
|
||||
published_date = cve.get("published")
|
||||
last_modified = cve.get("lastModified")
|
||||
|
||||
metrics = cve.get("metrics", {})
|
||||
cvss_v3 = metrics.get("cvssMetricV31", [])
|
||||
if cvss_v3:
|
||||
m = cvss_v3[0]
|
||||
cvss_data = m.get("cvssData", {})
|
||||
severity = cvss_data.get("baseSeverity", None)
|
||||
|
||||
records.append((name, version, cve_id, desc_en, severity, published_date, last_modified, True))
|
||||
|
||||
query = """
|
||||
INSERT INTO software_vulnerabilities (name, version, cve_id, description, severity, published_date, last_modified, has_vulnerabilities)
|
||||
VALUES %s
|
||||
ON CONFLICT (name, version, cve_id) DO UPDATE SET
|
||||
description = EXCLUDED.description,
|
||||
severity = EXCLUDED.severity,
|
||||
published_date = EXCLUDED.published_date,
|
||||
last_modified = EXCLUDED.last_modified,
|
||||
has_vulnerabilities = EXCLUDED.has_vulnerabilities,
|
||||
inserted_at = NOW();
|
||||
"""
|
||||
with conn.cursor() as cur:
|
||||
execute_values(cur, query, records)
|
||||
conn.commit()
|
||||
|
||||
else:
|
||||
# Нет уязвимостей - вставляем/обновляем NO_VULN
|
||||
query = """
|
||||
INSERT INTO software_vulnerabilities (name, version, cve_id, has_vulnerabilities)
|
||||
VALUES (%s, %s, 'NO_VULN', FALSE)
|
||||
ON CONFLICT (name, version, cve_id) DO UPDATE SET
|
||||
has_vulnerabilities = EXCLUDED.has_vulnerabilities,
|
||||
inserted_at = NOW(),
|
||||
description = NULL,
|
||||
severity = NULL,
|
||||
published_date = NULL,
|
||||
last_modified = NULL;
|
||||
"""
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(query, (name, version))
|
||||
conn.commit()
|
||||
|
||||
def main():
|
||||
# Подключаемся к БД
|
||||
conn = psycopg2.connect(
|
||||
host=DB_HOST,
|
||||
port=DB_PORT,
|
||||
dbname=DB_NAME,
|
||||
user=DB_USER,
|
||||
password=DB_PASSWORD
|
||||
)
|
||||
|
||||
jenkins_version = get_jenkins_version()
|
||||
if not jenkins_version:
|
||||
print("Could not determine Jenkins version.")
|
||||
sys.exit(1)
|
||||
|
||||
# Обрабатываем Jenkins Core
|
||||
jenkins_cpe = f"cpe:2.3:a:jenkins:jenkins:{jenkins_version}:*:*:*:-:*:*:*"
|
||||
print(f"Jenkins version: {jenkins_version}")
|
||||
print(f"Jenkins CPE: {jenkins_cpe}")
|
||||
jenkins_vulns = fetch_cves_by_cpe(jenkins_cpe)
|
||||
insert_vulnerabilities(conn, "jenkins_core", jenkins_version, jenkins_vulns)
|
||||
|
||||
# Обрабатываем плагины
|
||||
plugins = get_jenkins_plugins()
|
||||
print(f"\nFound {len(plugins)} plugins installed.")
|
||||
|
||||
for shortName, version in plugins:
|
||||
plugin_cpe = f"cpe:2.3:a:jenkins:{shortName}:{version}:*:*:*:*:jenkins:*:*"
|
||||
print(f"\nPlugin: {shortName}, version: {version}")
|
||||
print(f"CPE: {plugin_cpe}")
|
||||
plugin_vulns = fetch_cves_by_cpe(plugin_cpe)
|
||||
|
||||
# Добавляем суффикс _plugin для отличия в графиках
|
||||
plugin_name = f"{shortName}_plugin"
|
||||
insert_vulnerabilities(conn, plugin_name, version, plugin_vulns)
|
||||
|
||||
conn.close()
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user