vuln alerts from arch-audit

misc/checks.py

@@ -1,5 +1,5 @@
 from alerting import alerts
-from misc import sensors
+from misc import sensors, vuln
 
 IS_TESTING = False
 
@@ -65,3 +65,30 @@ def ram_check() -> list[alerts.Alert]:
     else:
         return []
     return [alert]
+
+
+async def vuln_check() -> list[alerts.Alert]:
+    vulns = await vuln.get_vulns()
+    alert_list = []
+    for v in vulns:
+        if IS_TESTING or v.fixed or v.severity in (vuln.Severity.HIGH, vuln.Severity.CRITICAL):
+            match v.severity:
+                case vuln.Severity.LOW:
+                    severity = alerts.Severity.INFO
+                case vuln.Severity.MEDIUM:
+                    severity = alerts.Severity.WARNING
+                case vuln.Severity.HIGH | vuln.Severity.CRITICAL:
+                    severity = alerts.Severity.CRITICAL
+            message = f"{v.id}: {v.vuln_type} in {','.join(v.packages)}"
+            html_message = f"<a href='{v.link}'>{v.id}</a>: {v.vuln_type} in {','.join(v.packages)}"
+            if v.fixed:
+                message.append(f" -- update to {v.fixed} ASAP")
+                html_message.append(f" -- update to {v.fixed} ASAP")
+            alert = alerts.Alert(
+                alert_type=alerts.AlertType.VULN,
+                message=message,
+                html_message=html_message,
+                severity=severity,
+            )
+            alert_list.append(alert)
+    return alert_list

misc/vuln.py

@@ -0,0 +1,55 @@
+import json
+import subprocess
+from dataclasses import dataclass
+from enum import StrEnum
+from typing import Optional
+
+from alerting import alerts
+
+
+class Severity(StrEnum):
+    LOW = "Low"
+    MEDIUM = "Medium"
+    HIGH = "High"
+    CRITICAL = "Critical"
+
+
+@dataclass
+class Vulnerability:
+    id: str
+    link: str
+    vuln_type: str
+    packages: list[str]
+    severity: Severity
+    fixed: Optional[str]
+
+
+def _parse_arch_audit_output(output: str) -> list[Vulnerability]:
+    arch_audit_json = json.loads(output)
+    vulnerabilities = []
+    for v in arch_audit_json:
+        vulnerability = Vulnerability(
+            id=v["name"],
+            link=f"https://security.archlinux.org/{v['name']}",
+            vuln_type=v["type"],
+            packages=v["packages"],
+            severity=v["severity"],
+            fixed=v["fixed"],
+        )
+        vulnerabilities.append(vulnerability)
+    return vulnerabilities
+
+
+async def get_vulns() -> list[Vulnerability]:
+    try:
+        arch_audit_output = subprocess.check_output(["arch-audit", "--json"])
+    except FileNotFoundError:
+        await alerts.send_alert(
+            alerts.Alert(
+                alert_type=alerts.AlertType.ERROR,
+                message="arch-audit not installed!",
+                severity=alerts.Severity.CRITICAL,
+            )
+        )
+        return []
+    return _parse_arch_audit_output(arch_audit_output)