395 lines
15 KiB
Python
395 lines
15 KiB
Python
import requests
|
||
import sys
|
||
import time
|
||
import os
|
||
import requests
|
||
import re
|
||
import psycopg2
|
||
from psycopg2.extras import execute_values
|
||
|
||
JENKINS_URL = "https://jenkins.lmru.tech"
|
||
AUTH_USER = "lm-sa-devsecops"
|
||
AUTH_TOKEN = "11b0f9ff6c986b18f0c7fa6e8ae769b66c"
|
||
|
||
VAULT_URL = "https://vault.lmru.tech"
|
||
|
||
ARTIFACTORY_URL = "https://art.lmru.tech"
|
||
|
||
SONARQUBE_URL = "https://sonar.lmru.tech"
|
||
|
||
CROWD_URL = "https://crowd.lmru.tech"
|
||
|
||
GHE_URL = "https://github.lmru.tech"
|
||
|
||
ARGOCD_URL = "https://argocd.devops.lmru.tech"
|
||
|
||
GRAFANA_OBS_URL = "https://obs-grafana-yc-techno.apps.lmru.tech"
|
||
|
||
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 get_vault_version():
|
||
"""Получить версию Vault."""
|
||
# Предполагаем, что Vault доступен по https://vault.lmru.tech/v1/sys/health
|
||
# и возвращает JSON с полем "version"
|
||
url = f"{VAULT_URL}/v1/sys/health"
|
||
resp = requests.get(url, verify=False)
|
||
resp.raise_for_status()
|
||
data = resp.json()
|
||
vault_version = data.get("version")
|
||
return vault_version
|
||
|
||
def get_artifactory_version():
|
||
# Artifactory version endpoint: /api/system/version
|
||
url = f"{ARTIFACTORY_URL}/api/system/version"
|
||
resp = requests.get(url, verify=False)
|
||
resp.raise_for_status()
|
||
data = resp.json()
|
||
artifactory_version = data.get("version")
|
||
return artifactory_version
|
||
|
||
def get_sonarqube_version():
|
||
# Эндпоинт SonarQube: /api/server/version
|
||
url = f"{SONARQUBE_URL}/api/server/version"
|
||
resp = requests.get(url, verify=False)
|
||
resp.raise_for_status()
|
||
# Версия приходит как простой текст (str)
|
||
sonarqube_version = resp.text.strip()
|
||
return sonarqube_version
|
||
|
||
def get_crowd_version():
|
||
url = f"{CROWD_URL}/crowd/console/login.action"
|
||
resp = requests.get(url, verify=False)
|
||
resp.raise_for_status()
|
||
html = resp.text
|
||
|
||
# Ищем значение атрибута value у input с id="crowd-version"
|
||
match_input = re.search(r'<input[^>]*id="crowd-version"[^>]*value="([^"]+)"', html)
|
||
if not match_input:
|
||
print("Could not find the crowd-version input field.")
|
||
return None
|
||
|
||
value_str = match_input.group(1) # например: 'Version: 4.2.2 (Build:#1581 - 2020-11-19) '
|
||
# Заменим на пробел
|
||
value_str = value_str.replace(' ', ' ')
|
||
|
||
# Ищем версию после слова Version:
|
||
match_version = re.search(r'Version:\s*([0-9]+\.[0-9]+\.[0-9]+)', value_str)
|
||
if match_version:
|
||
return match_version.group(1)
|
||
else:
|
||
print("Could not extract Crowd version from the value string.")
|
||
return None
|
||
|
||
def get_ghe_version():
|
||
url = f"{GHE_URL}/api/v3/meta"
|
||
resp = requests.get(url, verify=False)
|
||
resp.raise_for_status()
|
||
data = resp.json()
|
||
ghe_version = data.get("installed_version")
|
||
if ghe_version:
|
||
return ghe_version
|
||
else:
|
||
print("Could not extract GitHub Enterprise version from the /meta endpoint.")
|
||
return None
|
||
|
||
def get_argocd_version():
|
||
url = f"{ARGOCD_URL}/api/version"
|
||
resp = requests.get(url, verify=False)
|
||
resp.raise_for_status()
|
||
data = resp.json()
|
||
argocd_version = data.get("Version")
|
||
if argocd_version:
|
||
# Удалим префикс 'v'
|
||
argocd_version = argocd_version.lstrip('v')
|
||
# Если есть суффикс вида +коммита, уберём всё после +
|
||
if '+' in argocd_version:
|
||
argocd_version = argocd_version.split('+')[0]
|
||
return argocd_version
|
||
else:
|
||
print("Could not determine ArgoCD version from /api/version endpoint.")
|
||
return None
|
||
|
||
def get_oncall_version():
|
||
url = f"{GRAFANA_OBS_URL}/metrics"
|
||
resp = requests.get(url, verify=False)
|
||
resp.raise_for_status()
|
||
text = resp.text
|
||
|
||
# Ищем строку с plugin_id="grafana-oncall-app"
|
||
# Пример строки:
|
||
# grafana_plugin_build_info{plugin_id="grafana-oncall-app",plugin_type="app",signature_status="valid",version="1.3.58"} 1
|
||
match_line = re.search(r'grafana_plugin_build_info\{[^}]*plugin_id="grafana-oncall-app"[^}]*\}', text)
|
||
if not match_line:
|
||
print("Could not find grafana-oncall-app plugin info in /metrics.")
|
||
return None
|
||
|
||
line = match_line.group(0)
|
||
# Извлекаем version="..."
|
||
match_version = re.search(r'version="([^"]+)"', line)
|
||
if match_version:
|
||
oncall_version = match_version.group(1)
|
||
return oncall_version
|
||
else:
|
||
print("Could not extract version from grafana-oncall-app plugin line.")
|
||
return None
|
||
|
||
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, component, name, version, vulnerabilities):
|
||
"""
|
||
Вставить или обновить информацию об уязвимостях в PostgreSQL.
|
||
component - строка, например 'jenkins' или 'vault'.
|
||
"""
|
||
if vulnerabilities:
|
||
delete_query = """
|
||
DELETE FROM software_vulnerabilities
|
||
WHERE component = %s AND name = %s AND version = %s AND cve_id = 'NO_VULN';
|
||
"""
|
||
with conn.cursor() as cur:
|
||
cur.execute(delete_query, (component, 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)
|
||
|
||
# Добавляем component как первый элемент кортежа
|
||
records.append((component, name, version, cve_id, desc_en, severity, published_date, last_modified, True))
|
||
|
||
query = """
|
||
INSERT INTO software_vulnerabilities (component, name, version, cve_id, description, severity, published_date, last_modified, has_vulnerabilities)
|
||
VALUES %s
|
||
ON CONFLICT (component, 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:
|
||
query = """
|
||
INSERT INTO software_vulnerabilities (component, name, version, cve_id, has_vulnerabilities)
|
||
VALUES (%s, %s, %s, 'NO_VULN', FALSE)
|
||
ON CONFLICT (component, 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, (component, 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
|
||
component = "jenkins"
|
||
jenkins_version = get_jenkins_version()
|
||
if not jenkins_version:
|
||
print("Could not determine Jenkins version.")
|
||
sys.exit(1)
|
||
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, component, "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_name = f"{shortName}_plugin"
|
||
# # Для плагинов также component = 'jenkins', так как они относятся к Jenkins
|
||
# insert_vulnerabilities(conn, component, plugin_name, version, plugin_vulns)
|
||
|
||
# Vault
|
||
vault_version = get_vault_version()
|
||
if not vault_version:
|
||
print("Could not determine Vault version.")
|
||
sys.exit(1)
|
||
vault_version_clean = vault_version.split('+')[0]
|
||
vault_cpe = f"cpe:2.3:a:hashicorp:vault:{vault_version_clean}:*:*:*:enterprise:*:*:*"
|
||
print(f"Vault version: {vault_version}")
|
||
print(f"Vault CPE: {vault_cpe}")
|
||
# Для Vault component = 'vault'
|
||
vault_vulns = fetch_cves_by_cpe(vault_cpe)
|
||
insert_vulnerabilities(conn, "vault", "vault", vault_version, vault_vulns)
|
||
|
||
# Artifactory
|
||
artifactory_version = get_artifactory_version()
|
||
if not artifactory_version:
|
||
print("Could not determine Artifactory version.")
|
||
sys.exit(1)
|
||
artifactory_cpe = f"cpe:2.3:a:jfrog:artifactory:{artifactory_version}:*:*:*:*:*:*:*"
|
||
print(f"Artifactory version: {artifactory_version}")
|
||
print(f"Artifactory CPE: {artifactory_cpe}")
|
||
artifactory_vulns = fetch_cves_by_cpe(artifactory_cpe)
|
||
insert_vulnerabilities(conn, "artifactory", "artifactory", artifactory_version, artifactory_vulns)
|
||
|
||
# SonarQube
|
||
sonarqube_version = get_sonarqube_version()
|
||
if not sonarqube_version:
|
||
print("Could not determine SonarQube version.")
|
||
sys.exit(1)
|
||
sonarqube_cpe = f"cpe:2.3:a:sonarsource:sonarqube:{sonarqube_version}:*:*:*:*:*:*:*"
|
||
print(f"SonarQube version: {sonarqube_version}")
|
||
print(f"SonarQube CPE: {sonarqube_cpe}")
|
||
sonarqube_vulns = fetch_cves_by_cpe(sonarqube_cpe)
|
||
insert_vulnerabilities(conn, "sonarqube", "sonarqube", sonarqube_version, sonarqube_vulns)
|
||
|
||
# Crowd
|
||
crowd_version = get_crowd_version()
|
||
if not crowd_version:
|
||
print("Could not determine Crowd version.")
|
||
sys.exit(1)
|
||
crowd_cpe = f"cpe:2.3:a:atlassian:crowd:{crowd_version}:*:*:*:*:*:*:*"
|
||
print(f"Crowd version: {crowd_version}")
|
||
print(f"Crowd CPE: {crowd_cpe}")
|
||
crowd_vulns = fetch_cves_by_cpe(crowd_cpe)
|
||
insert_vulnerabilities(conn, "crowd", "crowd", crowd_version, crowd_vulns)
|
||
|
||
# GitHub Enterprise
|
||
ghe_version = get_ghe_version()
|
||
if not ghe_version:
|
||
print("Could not determine GitHub Enterprise version.")
|
||
sys.exit(1)
|
||
# Предположим такой CPE для GitHub Enterprise.
|
||
# Если точного CPE нет, можно поискать в NVD. Допустим:
|
||
# cpe:2.3:a:github:enterprise_server:<version>:*:*:*:*:*:*:*
|
||
ghe_cpe = f"cpe:2.3:a:github:enterprise_server:{ghe_version}:*:*:*:*:*:*:*"
|
||
print(f"GitHub Enterprise version: {ghe_version}")
|
||
print(f"GitHub Enterprise CPE: {ghe_cpe}")
|
||
ghe_vulns = fetch_cves_by_cpe(ghe_cpe)
|
||
insert_vulnerabilities(conn, "github_enterprise", "github_enterprise", ghe_version, ghe_vulns)
|
||
|
||
# ArgoCD
|
||
argocd_version = get_argocd_version()
|
||
if not argocd_version:
|
||
print("Could not determine ArgoCD version.")
|
||
sys.exit(1)
|
||
# Формируем CPE для ArgoCD:
|
||
argocd_cpe = f"cpe:2.3:a:argoproj:argo-cd:{argocd_version}:*:*:*:*:*:*:*"
|
||
print(f"ArgoCD version: {argocd_version}")
|
||
print(f"ArgoCD CPE: {argocd_cpe}")
|
||
|
||
argocd_vulns = fetch_cves_by_cpe(argocd_cpe)
|
||
insert_vulnerabilities(conn, "argocd", "argocd", argocd_version, argocd_vulns)
|
||
|
||
# OnCall
|
||
oncall_version = get_oncall_version()
|
||
if not oncall_version:
|
||
print("Could not determine OnCall version.")
|
||
sys.exit(1)
|
||
oncall_cpe = f"cpe:2.3:a:grafana:oncall:{oncall_version}:*:*:*:*:*:*:*"
|
||
print(f"OnCall version: {oncall_version}")
|
||
print(f"OnCall CPE: {oncall_cpe}")
|
||
|
||
oncall_vulns = fetch_cves_by_cpe(oncall_cpe)
|
||
insert_vulnerabilities(conn, "grafana", "oncall", oncall_version, oncall_vulns)
|
||
|
||
conn.close()
|
||
|
||
if __name__ == "__main__":
|
||
main() |