Merge pull request #25 from wolfSSL/fix/advisory-path-traversal

fix(advisory): validate ids used in output paths (path traversal)

Author: sameehj

central/gen-advisory

@@ -34,6 +34,7 @@ import argparse
 import json
 import os
 import pathlib
+import re
 import sys
 import urllib.request
 import uuid
@@ -43,6 +44,25 @@ from datetime import datetime, timezone
 GEN_TOOL_NAME = 'wolfssl-advisory-gen'
 GEN_TOOL_VERSION = '0.3'
 
+# CVE ids and advisory ids are interpolated into output filenames
+# (<id>.csaf.json / <id>.cdx.json) and into the CSAF self URL.  Records are
+# fetched from a remote API and parsed as arbitrary JSON, so an unvalidated id
+# is an untrusted-input -> arbitrary-file-write vector (e.g. cveId '../x').
+# Constrain both to a safe grammar with no path separators before either is
+# used to build a path.
+_CVE_ID_RE = re.compile(r'^CVE-[0-9]{4}-[0-9]{4,}$')
+_ADVISORY_ID_RE = re.compile(r'^[A-Za-z0-9][A-Za-z0-9._-]*$')
+
+
+def _validate_path_id(value, kind, pattern):
+    """Reject an id that is unsafe to interpolate into an output path."""
+    if not isinstance(value, str):
+        sys.exit(f"ERROR: refusing unsafe {kind}: expected string id, got {type(value).__name__}")
+    if not pattern.fullmatch(value):
+        sys.exit(f"ERROR: refusing unsafe {kind} {value!r}: must match "
+                 f"{pattern.pattern} (no path separators)")
+    return value
+
 _SCRIPTS_DIR = pathlib.Path(__file__).resolve().parent
 _REPO_ROOT = _SCRIPTS_DIR.parent
 
@@ -278,6 +298,7 @@ def parse_record(record):
     cve_id = meta.get('cveId') or cna.get('cveId')
     if not cve_id:
         sys.exit("ERROR: CVE record has no cveId")
+    _validate_path_id(cve_id, 'cveId', _CVE_ID_RE)
 
     description = ''
     for d in cna.get('descriptions', []):
@@ -878,6 +899,10 @@ def main():
                         'instead of batch mode.')
     args = p.parse_args()
 
+    # --advisory-id is interpolated into output filenames; constrain it too.
+    if args.advisory_id:
+        _validate_path_id(args.advisory_id, '--advisory-id', _ADVISORY_ID_RE)
+
     # ---- resolve the input records ----
     explicit = bool(args.cve_record or args.cve_id)
     if explicit:

central/test_gen_advisory.py

@@ -720,5 +720,67 @@ def test_malformed_record_fails_without_writing(self):
             self.assertFalse(os.path.exists(csaf))
 
 
+class TestPathIdValidation(unittest.TestCase):
+    """cveId and --advisory-id are interpolated into output filenames; a
+    record is fetched from a remote API and parsed as arbitrary JSON, so an
+    unvalidated id is an arbitrary-file-write vector.  Guard both."""
+
+    def _run(self, args):
+        return subprocess.run([sys.executable, str(SCRIPT)] + args,
+                              capture_output=True, text=True)
+
+    @staticmethod
+    def _record(cve_id):
+        return {'cveMetadata': {'cveId': cve_id},
+                'containers': {'cna': {
+                    'descriptions': [{'lang': 'en', 'value': 'test desc'}],
+                    'affected': [{'vendor': 'wolfSSL', 'product': 'wolfSSL',
+                                  'versions': []}]}}}
+
+    def test_traversal_cveid_rejected_and_writes_nothing_outside(self):
+        with tempfile.TemporaryDirectory() as d:
+            rec = os.path.join(d, 'evil.json')
+            with open(rec, 'w') as f:
+                json.dump(self._record('../ESCAPED'), f)
+            out = os.path.join(d, 'out', 'batch')
+            os.makedirs(out)
+            r = self._run(['--cve-record', rec, '--out-dir', out])
+            self.assertNotEqual(r.returncode, 0, r.stdout)
+            self.assertIn('unsafe cveId', r.stderr)
+            # The escaped path (sibling of out/, i.e. d/out/ESCAPED.*) must
+            # not have been written.
+            escaped = os.path.join(d, 'out', 'ESCAPED.csaf.json')
+            self.assertFalse(os.path.exists(escaped), escaped)
+
+    def test_absolute_cveid_rejected(self):
+        with tempfile.TemporaryDirectory() as d:
+            rec = os.path.join(d, 'abs.json')
+            with open(rec, 'w') as f:
+                json.dump(self._record('/etc/pwned'), f)
+            r = self._run(['--cve-record', rec, '--out-dir', d])
+            self.assertNotEqual(r.returncode, 0, r.stdout)
+            self.assertIn('unsafe cveId', r.stderr)
+
+    def test_well_formed_cveid_accepted(self):
+        with tempfile.TemporaryDirectory() as d:
+            rec = os.path.join(d, 'good.json')
+            with open(rec, 'w') as f:
+                json.dump(self._record('CVE-2026-12345'), f)
+            r = self._run(['--cve-record', rec, '--out-dir', d])
+            self.assertEqual(r.returncode, 0, r.stderr)
+            self.assertTrue(
+                os.path.exists(os.path.join(d, 'CVE-2026-12345.csaf.json')))
+
+    def test_traversal_advisory_id_rejected(self):
+        with tempfile.TemporaryDirectory() as d:
+            rec = os.path.join(d, 'good.json')
+            with open(rec, 'w') as f:
+                json.dump(self._record('CVE-2026-12345'), f)
+            r = self._run(['--cve-record', rec, '--out-dir', d,
+                           '--advisory-id', '../evil'])
+            self.assertNotEqual(r.returncode, 0, r.stdout)
+            self.assertIn('unsafe --advisory-id', r.stderr)
+
+
 if __name__ == '__main__':
     unittest.main()