Skip to content

Commit 4447af8

Browse files
jdstrandjdstrand
committed
feat: implement new cve-specific rfc5322 parser
Python 3.12 changed the email.parser parsing as part of python/cpython#124452, but this broke our parser. We were using email.parser as a cheap way to parse RFC5322, but we're already had to do post-processing for our format and would need to add compatibility for the new parsing. Instead, implement our own RFC5322 parser and tailor it to our needs (which allows better formatting checks).
1 parent 934eb8d commit 4447af8

4 files changed

Lines changed: 678 additions & 68 deletions

File tree

‎cvelib/common.py‎

Lines changed: 6 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -13,9 +13,7 @@
1313
import time
1414
from typing import Any, Dict, List, Optional, Pattern, Set, Tuple, Union
1515

16-
from email.message import Message
17-
from email.parser import HeaderParser, Parser
18-
from email.policy import Compat32
16+
from cvelib.rfc5322 import parseHeaders
1917

2018

2119
# cache of config
@@ -459,52 +457,18 @@ def readFile(fn: str) -> Optional[Set[str]]:
459457

460458
def readCve(fn: str) -> Dict[str, str]:
461459
"""Read raw CVE data from file"""
462-
# Read in the data, but let callers do any specific formatting
463-
d: Dict[str, str] = {}
464-
465460
# Use a relative filename for warnings
466461
rel_fn: str = os.path.basename(fn)
467462
parent: str = os.path.basename(os.path.dirname(fn))
468463
if parent in cve_reldirs:
469464
rel_fn = "%s/%s" % (parent, rel_fn)
470465

471-
# Encode to to utf-8 since we do isprintable() checks elsewhere
472466
with open(fn, "r", encoding="utf-8", errors="backslashreplace") as fp:
473-
policy: Compat32 = Compat32()
474-
475-
# Obtain the header content
476-
headers: Message = Parser(policy=policy).parse(fp)
477-
for k in headers:
478-
if k in d:
479-
warn("duplicate key '%s' in %s" % (k, rel_fn))
480-
d[k] = headers[k]
481-
482-
# Obtain the header content for any other stanzas (stanzas are
483-
# separated by newlines so we need to grab the stanza'a headers from
484-
# the payload in a loop.
485-
last: Optional[str] = None
486-
while True:
487-
s: str = headers.get_payload()
488-
if s == last:
489-
break
490-
last = s
491-
headers: Message = HeaderParser(policy=policy).parsestr(s)
492-
for k in headers:
493-
if k in d:
494-
warn("duplicate key '%s' in %s" % (k, rel_fn))
495-
d[k] = headers[k]
496-
497-
return copy.deepcopy(d)
498-
499-
500-
def setCveHeader(headers: Message, key: str, val: Optional[str]) -> None:
501-
"""Set header for CVE"""
502-
if val is None:
503-
headers.__delitem__(key) # no exception if missing
504-
elif key in headers:
505-
headers.replace_header(key, val)
506-
else:
507-
headers.add_header(key, val)
467+
content: str = fp.read()
468+
469+
return copy.deepcopy(
470+
parseHeaders(content, warnFn=lambda msg: warn("%s in %s" % (msg, rel_fn)))
471+
)
508472

509473

510474
def getCacheDirPath() -> str:

‎cvelib/rfc5322.py‎

Lines changed: 159 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,159 @@
1+
#!/usr/bin/env python3
2+
#
3+
# SPDX-License-Identifier: MIT
4+
5+
"""RFC5322-style header parser.
6+
7+
This module provides a simple parser for RFC5322-style headers (the format
8+
used in email headers). Historically, sedg used the email.parser module but
9+
to avoid Python version-specific behavior changes in how multiline header
10+
values are parsed, we now implement our own parser.
11+
"""
12+
13+
import re
14+
from typing import Callable, Dict, List, Optional
15+
16+
# RFC5322 field name: printable ASCII except colon and space
17+
# We extend this to allow characters commonly used in this codebase:
18+
# - hyphens (e.g., "Discovered-by")
19+
# - underscores (e.g., "git_pkg1")
20+
# - forward slashes (e.g., "git/org_foo", "snap/pub_pkg2")
21+
# - dots (e.g., "oci/gar-us.proj_pkg1")
22+
# - hash/pound at start (e.g., "#Patches_PKG:" for commented-out headers)
23+
#
24+
# Notes:
25+
# - email.parser accepts these, downstream validation rejects invalid keys
26+
# - RFC5322 has a concept of an email body, but the Cve format only uses
27+
# headers, so anything that looks like a body will raise an error.
28+
_VALID_KEY_PATTERN = re.compile(r"^#?[A-Za-z][A-Za-z0-9_./-]*$")
29+
30+
# Maximum limits to prevent DoS
31+
MAX_LINE_LENGTH = 65536
32+
MAX_LINES = 100000
33+
MAX_KEY_LENGTH = 256
34+
MAX_VALUE_LENGTH = 1048576 # 1MB
35+
36+
37+
class Rfc5322Exception(Exception):
38+
"""This class represents RFC5322 parsing exceptions"""
39+
40+
def __init__(self, value: str) -> None:
41+
self.value = value
42+
43+
44+
def parseHeaders(
45+
content: str,
46+
warnFn: Optional[Callable[[str], None]] = None,
47+
) -> Dict[str, str]:
48+
"""Parse RFC5322-style headers.
49+
50+
Args:
51+
content: The content to parse
52+
warnFn: Optional callback for warnings (e.g., duplicate keys)
53+
54+
Returns:
55+
Dictionary mapping header names to values. For multiline values
56+
(continuation lines starting with space/tab), the value is prefixed
57+
with a newline character.
58+
59+
Raises:
60+
Rfc5322Exception: If content exceeds safety limits or contains
61+
malformed lines (lines without colons that aren't continuations)
62+
"""
63+
result: Dict[str, str] = {}
64+
current_key: Optional[str] = None
65+
current_lines: List[str] = []
66+
is_multiline: bool = False
67+
68+
def save_current() -> None:
69+
nonlocal current_key, current_lines, is_multiline
70+
if current_key:
71+
if is_multiline:
72+
val = "\n" + "\n".join(current_lines)
73+
else:
74+
val = current_lines[0] if current_lines else ""
75+
# Check value length limit
76+
if len(val) > MAX_VALUE_LENGTH:
77+
raise Rfc5322Exception(
78+
"value for '%s' exceeds maximum length" % current_key
79+
)
80+
if current_key in result:
81+
if warnFn:
82+
warnFn("duplicate key '%s'" % current_key)
83+
result[current_key] = val
84+
current_key = None
85+
current_lines = []
86+
is_multiline = False
87+
88+
# Normalize line endings (CRLF -> LF)
89+
content = content.replace("\r\n", "\n").replace("\r", "\n")
90+
91+
# Handle empty content
92+
if not content:
93+
return result
94+
95+
lines = content.split("\n")
96+
97+
# Check line count limit
98+
if len(lines) > MAX_LINES:
99+
raise Rfc5322Exception("content exceeds maximum line count")
100+
101+
# Parse the content looking for key and single line or multiline values.
102+
# When a key is found, record it as current_key and its value is appended
103+
# to current_lines (for single of multi lines). On empty line or finished
104+
# processing multiline, save the current_key and its value (current_lines)
105+
# in results.
106+
for line in lines:
107+
# Check line length limit
108+
if len(line) > MAX_LINE_LENGTH:
109+
raise Rfc5322Exception("line exceeds maximum length")
110+
111+
if not line:
112+
# Empty line - stanza separator (but not allowed before first header)
113+
if not current_key and not result:
114+
raise Rfc5322Exception("leading blank line")
115+
save_current()
116+
continue
117+
118+
if line[0] in [" ", "\t"]:
119+
# Continuation line - must have a current key
120+
if not current_key:
121+
raise Rfc5322Exception(
122+
"continuation line without preceding header: '%s'" % line[:50]
123+
)
124+
current_lines.append(line)
125+
is_multiline = True
126+
continue
127+
128+
if ":" not in line:
129+
raise Rfc5322Exception("line without colon: '%s'" % line[:50])
130+
131+
# New header - save previous first
132+
save_current()
133+
134+
key, _, value = line.partition(":")
135+
136+
# Validate key
137+
if not key:
138+
raise Rfc5322Exception("empty key (line starts with ':')")
139+
140+
if len(key) > MAX_KEY_LENGTH:
141+
raise Rfc5322Exception("key exceeds maximum length: '%s'" % key[:50])
142+
143+
if not _VALID_KEY_PATTERN.match(key):
144+
raise Rfc5322Exception("invalid key: '%s'" % key[:50])
145+
146+
current_key = key
147+
148+
# Warn if no space after colon for non-empty single-line values
149+
if value and not value[0].isspace():
150+
if warnFn:
151+
warnFn("missing space after colon for key '%s'" % key)
152+
153+
value = value.lstrip()
154+
current_lines = [value] if value else []
155+
156+
# Handle last header
157+
save_current()
158+
159+
return result

‎tests/test_common.py‎

Lines changed: 26 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -3,12 +3,12 @@
33
#
44
# SPDX-License-Identifier: MIT
55

6-
from email.message import EmailMessage
76
import json
87
import os
98
from unittest import TestCase, mock, skipIf
109

1110
import cvelib.common
11+
import cvelib.rfc5322
1212
import tests.testutil
1313

1414

@@ -103,27 +103,6 @@ def test_epochToISO8601(self):
103103
with self.assertRaises(ValueError):
104104
cvelib.common.epochToISO8601(input)
105105

106-
def test_setCveHeader(self):
107-
"""Test setCveHeader()"""
108-
m = EmailMessage()
109-
self.assertEqual(0, len(m))
110-
111-
# add
112-
cvelib.common.setCveHeader(m, "foo", "bar")
113-
self.assertEqual(1, len(m))
114-
self.assertTrue("foo" in m)
115-
self.assertEqual("bar", m["foo"])
116-
117-
# replace
118-
cvelib.common.setCveHeader(m, "foo", "baz")
119-
self.assertEqual(1, len(m))
120-
self.assertTrue("foo" in m)
121-
self.assertEqual("baz", m["foo"])
122-
123-
# delete
124-
cvelib.common.setCveHeader(m, "foo", None)
125-
self.assertEqual(0, len(m))
126-
127106
def test_getCacheDirPath(self):
128107
"""Test getCacheDirPath()"""
129108
if "XDG_CACHE_HOME" in os.environ: # pragma: nocover
@@ -448,11 +427,8 @@ def test_readCVE(self):
448427
"dupe-test2: bar\nbaz: norf\n\ndupe-test2: bar\n",
449428
{"dupe-test2": "bar", "baz": "norf"},
450429
),
451-
# weird cases
452-
("bad", {}),
453-
("f\x00o: bar", {}),
430+
# values with unusual characters are fine
454431
("foo: b\xaar", {"foo": "b\xaar"}),
455-
("😀: bar", {}), # utf-8 F09F9880
456432
("foo: 😀", {"foo": "😀"}),
457433
]
458434
for inp, exp in tsts:
@@ -474,6 +450,30 @@ def test_readCVE(self):
474450
self.assertEqual("", output.getvalue().strip())
475451
self.assertEqual("", error.getvalue().strip())
476452

453+
# test malformed input raises exception
454+
fn = os.path.join(self.tmpdir, "testcve")
455+
456+
# line without colon
457+
with open(fn, "w") as f:
458+
f.write("bad")
459+
with self.assertRaises(cvelib.rfc5322.Rfc5322Exception) as ctx:
460+
cvelib.common.readCve(fn)
461+
self.assertEqual("line without colon: 'bad'", ctx.exception.value)
462+
463+
# invalid key with null byte
464+
with open(fn, "w") as f:
465+
f.write("f\x00o: bar")
466+
with self.assertRaises(cvelib.rfc5322.Rfc5322Exception) as ctx:
467+
cvelib.common.readCve(fn)
468+
self.assertEqual("invalid key: 'f\x00o'", ctx.exception.value)
469+
470+
# invalid key with emoji
471+
with open(fn, "w") as f:
472+
f.write("😀: bar")
473+
with self.assertRaises(cvelib.rfc5322.Rfc5322Exception) as ctx:
474+
cvelib.common.readCve(fn)
475+
self.assertEqual("invalid key: '😀'", ctx.exception.value)
476+
477477
def test_readFile(self):
478478
"""Test readFile()"""
479479
self.tmpdir = tests.testutil._createTmpDir()

0 commit comments

Comments
 (0)