VULN/vuln-suse-vex
1
0
Fork 0
Code Pull Requests Actions Activity

Initial SUSE VEX mirror

Browse Source
This commit is contained in:
root
2026-11-10 22:31:12 +01:00
commit 72c7e1ad41
9 changed files with 364 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitea/workflows/update.yml
+41
View File
@@ -0,0 +1,41 @@
name: SUSE VEX Mirror
on:
workflow_dispatch:
schedule:
- cron: "49 */6 * * *"
permissions:
contents: write
concurrency:
group: vuln-suse-vex-sync
cancel-in-progress: false
jobs:
mirror:
runs-on: vuln-runner
timeout-minutes: 360
steps:
- name: Sync SUSE VEX using persistent Git checkout
env:
GITEA_JOB_TOKEN: ${{ secrets.GITEA_TOKEN }}
CACHE_ROOT: /var/lib/act_runner/cache/vuln-suse-vex
REPO_URL: https://git.apex-project.de/VULN/vuln-suse-vex.git
run: |
set -Eeuo pipefail
umask 077
: "${GITEA_JOB_TOKEN:?Missing GITEA_TOKEN}"
mkdir -p "$CACHE_ROOT"
exec 9>"$CACHE_ROOT/.bootstrap.lock"
flock -w 300 9
AUTH_HEADER="Authorization: Basic $(printf 'oauth2:%s' "$GITEA_JOB_TOKEN" | base64 -w0)"
export GIT_CONFIG_COUNT=1
export GIT_CONFIG_KEY_0=http.extraheader
export GIT_CONFIG_VALUE_0="$AUTH_HEADER"
export GIT_TERMINAL_PROMPT=0
if [ ! -d "$CACHE_ROOT/repo/.git" ]; then
git clone --depth 1 --branch main "$REPO_URL" "$CACHE_ROOT/repo"
fi
unset AUTH_HEADER
bash "$CACHE_ROOT/repo/scripts/run-cached.sh"
.gitignore
+6
View File
@@ -0,0 +1,6 @@
__pycache__/
*.pyc
.pytest_cache/
*.tmp
.download-*
.env
README.md
+42
View File
@@ -0,0 +1,42 @@
# vuln-suse-vex
Eigener Gitea-Mirror der **SUSE CSAF 2.0 VEX**-Dokumente (je CVE) ab 2017.
Die bereits angelegte Gitea-Adresse ist `https://git.apex-project.de/VULN/vuln-suse-vex`.
## Quelle
- Offizielle SUSE-Quelle: `https://ftp.suse.com/pub/projects/security/csaf-vex.tar.bz2`
- Ausgabe: `data/vex/YYYY/CVE-YYYY-NNNN.json`
- Der Downloader verwendet `If-None-Match` und `If-Modified-Since` soweit die Quelle diese Header unterstützt.
- **Hinweis:** Bei einem geänderten Tar-Archiv wird das vollständige Archiv erneut geladen. Das ist kein echtes File-Delta. Das Repository selbst verwendet einen persistenten Git-Checkout.
- Kein Fallback auf möglicherweise veraltete Mirrors. Bei Fehlern bleibt der letzte Datenbestand unverändert.
## Erstinstallation
1. Leeres Gitea-Repository `VULN/vuln-suse-vex` anlegen.
2. Dateien committen und nach `main` pushen.
3. Auf der Gitea-VM:
```bash
sudo apt update
sudo apt install -y git python3 ca-certificates util-linux bzip2
sudo install -d -o act_runner -g act_runner -m 700 /var/lib/act_runner/cache/vuln-suse-vex
```
4. Actions aktivieren. Runner-Label `vuln-runner` verwenden. `secrets.GITEA_TOKEN` muss Schreibrechte besitzen.
5. Workflow `SUSE VEX Mirror` manuell starten; danach alle 6 Stunden zur Minute 49 UTC.
## Fehlerbehandlung
- Ein unterbrochener Import lässt möglicherweise `?? data/` zurück. Das Skript bewahrt diese Dateien und synchronisiert sie bei gleichbleibendem Remote-Commit erneut.
- Wenn sich `main` geändert hat **und** der Cache uncommittete Dateien enthält, wird aus Sicherheitsgründen abgebrochen. Keinesfalls `git clean -fd` oder `git reset --hard` auf diese Daten anwenden, ohne sie vorher zu prüfen.
- Eine geänderte Quelldatei wird zunächst vollständig heruntergeladen und erst nach erfolgreicher Archiv- und JSON-Validierung verarbeitet.
- Nach einem unterbrochenen Commit/Push lokale Git-Historie prüfen.
- Lokaler Test: `python3 -m unittest discover -s tests -v`.
## Grenzen
- Keine Integration von separaten SUSE-Advisories; diese liegen in `vuln-suse`.
- Der Feed kann größer als mehrere hundert MB sein; ausreichend Platz für Archiv, JSON-Dateien und Git-Historie vorsehen.
- `state/sync.json` wird für einen erfolgreichen Snapshot aktualisiert. Die zusätzliche HTTP-Metadatei bleibt nur auf der VM.
- Vor produktiver Freigabe einen kompletten Live-Import testen.
data/.gitkeep
View File
scripts/run-cached.sh
+51
View File
@@ -0,0 +1,51 @@
#!/usr/bin/env bash
set -Eeuo pipefail
umask 077
: "${GITEA_JOB_TOKEN:?GITEA_JOB_TOKEN missing}"
ROOT="${CACHE_ROOT:-/var/lib/act_runner/cache/vuln-suse-vex}"
REPO_URL="${REPO_URL:-https://git.apex-project.de/VULN/vuln-suse-vex.git}"
mkdir -p "$ROOT"
exec 9>"$ROOT/.sync.lock"
flock -w 300 9 || { echo 'Another sync is active' >&2; exit 1; }
cd "$ROOT"
# Credentials only provided to git commands in process environment; never saved to .git/config.
AUTH_HEADER="Authorization: Basic $(printf 'oauth2:%s' "$GITEA_JOB_TOKEN" | base64 -w0)"
export GIT_TERMINAL_PROMPT=0
GIT_CONFIG_COUNT=1
GIT_CONFIG_KEY_0=http.extraheader
GIT_CONFIG_VALUE_0="$AUTH_HEADER"
export GIT_CONFIG_COUNT GIT_CONFIG_KEY_0 GIT_CONFIG_VALUE_0
if [[ ! -d repo/.git ]]; then
git clone --depth 1 --branch main "$REPO_URL" repo
fi
cd repo
# Allow only partial data/state from interrupted jobs. Never discard automatically.
DIRTY=$(git status --porcelain)
if [[ -n "$DIRTY" ]]; then
if git status --porcelain | grep -Ev '^.. (data/|state/)' >/dev/null; then
echo 'Unexpected dirty paths; manual review required' >&2; exit 1
fi
echo 'Previous interrupted import data found; preserving it.'
fi
git fetch --depth 1 origin main
LOCAL=$(git rev-parse HEAD)
REMOTE=$(git rev-parse FETCH_HEAD)
if [[ "$LOCAL" != "$REMOTE" ]]; then
if [[ -n "$DIRTY" ]]; then
echo 'Upstream changed while data/state are dirty; manual recovery needed. No reset performed.' >&2
exit 1
fi
git reset --hard FETCH_HEAD
fi
python3 scripts/sync.py --since-year 2017
python3 scripts/validate.py
git config user.name 'Vulnerability Bot'
git config user.email 'vuln-bot@users.noreply.local'
git add -A -- data/ state/
if git diff --cached --quiet; then
echo 'No data changes'
exit 0
fi
git commit -m "Sync SUSE VEX $(date -u +%Y-%m-%dT%H:%M:%SZ)"
# Concurrent push can't overwrite others. On failure keep local commit for inspection.
git push origin HEAD:main
scripts/sync.py
+159
View File
@@ -0,0 +1,159 @@
#!/usr/bin/env python3
"""SUSE CSAF VEX mirror from official upstream tarball.
Atomic snapshot: download complete archive, validate every selected JSON before
changing the dataset, then reconcile data, including deletions. Resume: existing
uncommitted files are retained and compared against the next valid snapshot.
"""
import argparse
import hashlib
import json
import os
from pathlib import Path
import re
import shutil
import sys
import tarfile
import tempfile
import time
from datetime import datetime, timezone
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen
DEFAULT_URL = 'https://ftp.suse.com/pub/projects/security/csaf-vex.tar.bz2'
CVE_RE = re.compile(r'(?i)cve[-_](\d{4})[-_](\d{4,})')
MAX_MEMBER = 32 * 1024 * 1024
UA = 'vuln-suse-vex-mirror/1.0 (+https://git.apex-project.de/VULN/vuln-suse-vex)'
def atomic_write(path, data):
path.parent.mkdir(parents=True, exist_ok=True)
fd, tmp = tempfile.mkstemp(dir=str(path.parent), prefix='.tmp-')
try:
with os.fdopen(fd, 'wb') as f:
f.write(data)
f.flush()
os.fsync(f.fileno())
os.replace(tmp, path)
finally:
if os.path.exists(tmp):
os.unlink(tmp)
def fetch(url, archive, meta_path, timeout, retries):
meta = json.loads(meta_path.read_text()) if meta_path.exists() else {}
headers = {'User-Agent': UA}
if archive.exists() and archive.stat().st_size > 0 and meta.get('source') == url:
if meta.get('etag'): headers['If-None-Match'] = meta['etag']
if meta.get('last_modified'): headers['If-Modified-Since'] = meta['last_modified']
for n in range(retries):
try:
req = Request(url, headers=headers)
with urlopen(req, timeout=timeout) as response:
fd, tmp = tempfile.mkstemp(dir=str(archive.parent), prefix='.download-')
try:
with os.fdopen(fd, 'wb') as f:
shutil.copyfileobj(response, f, length=1024*1024)
f.flush(); os.fsync(f.fileno())
if os.stat(tmp).st_size < 1024:
raise ValueError('Archive unexpectedly small')
# Validate compressed tar before replacing persistent archive
with tarfile.open(tmp, 'r:bz2') as tf:
if tf.next() is None: raise ValueError('Empty upstream tar')
os.replace(tmp, archive)
new_meta = {'source':url, 'etag':response.headers.get('ETag'),
'last_modified':response.headers.get('Last-Modified')}
atomic_write(meta_path,json.dumps(new_meta, indent=2).encode())
return True
finally:
if os.path.exists(tmp):os.unlink(tmp)
except HTTPError as e:
if e.code == 304 and archive.exists():return False
if e.code not in (429, 500, 502, 503, 504):raise
err=e
except (OSError, URLError, ValueError, tarfile.TarError) as e:
err=e
if n+1 < retries:
delay=min(2**n * 8, 120)
print(f'Download failed: {err}; retry in {delay}s',file=sys.stderr,flush=True)
time.sleep(delay)
raise RuntimeError(f'Cannot fetch {url}: {err}')
def parse_archive(path, since_year):
"""Read and validate ALL relevant raw VEX JSON before any output mutation."""
items={}
with tarfile.open(path, 'r|bz2') as tf:
for member in tf:
if not member.isfile() or not member.name.lower().endswith('.json'):
continue
name=Path(member.name).name
match=CVE_RE.search(name)
if not match or int(match.group(1)) < since_year:
continue
if member.size <= 0 or member.size > MAX_MEMBER:
raise ValueError(f'Invalid JSON member size: {member.name}')
f=tf.extractfile(member)
if not f:raise ValueError(f'Unreadable member: {member.name}')
raw=f.read(MAX_MEMBER+1)
if len(raw) != member.size or len(raw)>MAX_MEMBER:
raise ValueError(f'Incomplete JSON: {member.name}')
obj=json.loads(raw)
if not isinstance(obj,dict) or not isinstance(obj.get('document'),dict) or not isinstance(obj.get('vulnerabilities'),list):
raise ValueError(f'Invalid CSAF document: {member.name}')
year=match.group(1)
cve=f'CVE-{year}-{match.group(2)}'
# Duplicate names across archive: only exact duplicate bytes are acceptable
target=f'{year}/{cve}.json'
if target in items and items[target] != raw:
raise ValueError(f'Duplicate conflicting CVE document: {target}')
items[target]=raw
if not items:raise ValueError('Zero CSAF VEX documents; refusing snapshot')
return items
def reconcile(items, root, state, source, since_year):
root.mkdir(parents=True,exist_ok=True)
existing={p.relative_to(root).as_posix():p for p in root.rglob('*.json')}
added=changed=removed=0
for rel,raw in items.items():
target=root/rel
if target.exists() and target.read_bytes()==raw:continue
added += not target.exists(); changed += target.exists()
atomic_write(target,raw)
for rel,p in existing.items():
# Files older than since_year never belong to this mirror
if rel not in items:
p.unlink();removed+=1
# This state describes a completely validated snapshot, not an incomplete run
payload={'source':source,'min_year':since_year,'record_count':len(items),
'snapshot_sha256':hashlib.sha256('\n'.join(sorted(items)).encode()).hexdigest(),
'completed_at':datetime.now(timezone.utc).isoformat(),
'added':added,'updated':changed,'deleted':removed}
atomic_write(state,json.dumps(payload,indent=2,sort_keys=True).encode())
return payload
def main():
ap=argparse.ArgumentParser()
ap.add_argument('--source',default=os.environ.get('SUSE_VEX_URL',DEFAULT_URL))
ap.add_argument('--since-year',type=int,default=2017)
ap.add_argument('--cache',type=Path,default=Path(os.environ.get('SUSE_VEX_DOWNLOAD_CACHE','/var/lib/act_runner/cache/vuln-suse-vex/download')))
ap.add_argument('--timeout',type=int,default=180)
ap.add_argument('--retries',type=int,default=4)
args=ap.parse_args()
if args.since_year<1999:ap.error('since-year must be >= 1999')
args.cache.mkdir(parents=True,exist_ok=True)
archive=args.cache/'csaf-vex.tar.bz2'
meta=args.cache/'http.json'
downloaded=fetch(args.source,archive,meta,args.timeout,args.retries)
# Don't rewrite the state, and don't regenerate a commit if HTTP returns 304.
if not downloaded and Path('state/sync.json').exists():
print('Upstream archive unchanged (HTTP 304)')
return
records=parse_archive(archive,args.since_year)
result=reconcile(records,Path('data/vex'),Path('state/sync.json'),args.source,args.since_year)
print(json.dumps(result,indent=2))
if __name__=='__main__':
main()
scripts/validate.py
+24
View File
@@ -0,0 +1,24 @@
#!/usr/bin/env python3
"""Fail closed on invalid CSAF JSON and incomplete synchronization."""
import json
from pathlib import Path
import re
import sys
CVE=re.compile(r'^CVE-\d{4}-\d{4,}\.json$',re.I)
def validate(root=Path('data/vex'), state=Path('state/sync.json')):
if not state.exists():raise ValueError('Missing completed sync state')
metadata=json.loads(state.read_text())
files=sorted(root.rglob('*.json'))
if not files or len(files)!=metadata['record_count']:
raise ValueError(f'Count mismatch: files={len(files)} expected={metadata["record_count"]}')
for path in files:
if not CVE.match(path.name):raise ValueError(f'Invalid filename: {path}')
doc=json.loads(path.read_bytes())
if not isinstance(doc,dict) or not isinstance(doc.get('document'),dict) or not isinstance(doc.get('vulnerabilities'),list):
raise ValueError(f'Invalid CSAF VEX: {path}')
print(f'Validated {len(files)} CSAF VEX documents')
if __name__=='__main__':
try:validate()
except (ValueError,OSError,KeyError,json.JSONDecodeError) as ex:
print(f'VALIDATION FAILED: {ex}',file=sys.stderr);sys.exit(1)
state/.gitkeep
View File
tests/test_sync.py
+41
View File
@@ -0,0 +1,41 @@
import bz2
import io
import json
import tarfile
import tempfile
import unittest
from pathlib import Path
import sys
sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'scripts'))
from sync import parse_archive,reconcile
from validate import validate
class SyncTests(unittest.TestCase):
def test_roundtrip_and_deletions(self):
with tempfile.TemporaryDirectory() as tmp:
p=Path(tmp);archive=p/'test.tar.bz2'
data=json.dumps({'document':{'category':'csaf_vex'},'vulnerabilities':[]}).encode()
with tarfile.open(archive,'w:bz2') as tf:
for name in ('csaf-vex/2025/cve-2025-12345.json','csaf-vex/2016/cve-2016-9999.json'):
info=tarfile.TarInfo(name);info.size=len(data);tf.addfile(info,io.BytesIO(data))
items=parse_archive(archive,2017)
self.assertEqual(list(items),['2025/CVE-2025-12345.json'])
import os
cwd=Path.cwd();os.chdir(p)
try:
result=reconcile(items,Path('data/vex'),Path('state/sync.json'),'test',2017)
self.assertEqual(result['record_count'],1)
validate()
(Path('data/vex/2025')/'CVE-2025-12345.json').unlink()
reconcile(items,Path('data/vex'),Path('state/sync.json'),'test',2017)
validate()
finally:os.chdir(cwd)
def test_invalid_csaf_fails_before_write(self):
with tempfile.TemporaryDirectory() as tmp:
p=Path(tmp);archive=p/'bad.tar.bz2'
data=b'{bad'
with tarfile.open(archive,'w:bz2') as tf:
x=tarfile.TarInfo('cve-2026-1111.json');x.size=len(data);tf.addfile(x,io.BytesIO(data))
with self.assertRaises(json.JSONDecodeError):parse_archive(archive,2017)
if __name__=='__main__':unittest.main()