diff --git a/__pycache__/parser.cpython-312.pyc b/__pycache__/parser.cpython-312.pyc new file mode 100644 index 0000000..275ec2c Binary files /dev/null and b/__pycache__/parser.cpython-312.pyc differ diff --git a/__pycache__/test_parser.cpython-312.pyc b/__pycache__/test_parser.cpython-312.pyc new file mode 100644 index 0000000..3531943 Binary files /dev/null and b/__pycache__/test_parser.cpython-312.pyc differ diff --git a/compliance.md b/compliance.md new file mode 100644 index 0000000..97473c2 --- /dev/null +++ b/compliance.md @@ -0,0 +1,66 @@ +# RFC 5322 Address Parser — Compliance Matrix + +Maps every ABNF production used in address parsing to its RFC section, +exercising test case(s), and implementation status. + +| ABNF production | RFC section | Test case(s) | Status | +|-----------------|-------------|--------------|--------| +| `quoted-pair` | §3.2.1 | `TestQuotedPair_3_2_1.*` | complete | +| `obs-qp` | §4.1 / §3.2.1 | permissive path in `_try_quoted_pair` | complete | +| `FWS` | §3.2.2 | `TestFWS_3_2_2.*` | complete | +| `obs-FWS` | §4.2 | covered via permissive FWS folding | partial | +| `ctext` | §3.2.2 | `TestCFWSComments_3_2_3.*` | complete | +| `obs-ctext` | §4.1 | permissive comment path | complete | +| `ccontent` | §3.2.2 | nested + quoted-pair comment tests | complete | +| `comment` | §3.2.2 | `test_nested_comments`, `test_leading_comment` | complete | +| `CFWS` | §3.2.2 | `TestCFWSComments_3_2_3.*`, FWS tests | complete | +| `atext` | §3.2.3 | `test_atext_specials_in_local`, simple addr-spec | complete | +| `atom` | §3.2.3 | phrase/display-name tests | complete | +| `dot-atom-text` | §3.2.3 | `test_dot_atom_local`, domain tests | complete | +| `dot-atom` | §3.2.3 | `test_simple_addr_spec`, domain tests | complete | +| `specials` | §3.2.3 | used as complement of atext (lexer) | complete | +| `qtext` | §3.2.4 | `TestQuotedString_3_2_4.*` | complete | +| `obs-qtext` | §4.1 | permissive quoted-string path | complete | +| `qcontent` | §3.2.4 | quoted-pair + qtext mix tests | complete | +| `quoted-string` | §3.2.4 | `TestQuotedString_3_2_4.*` | complete | +| `word` | §3.2.5 | `TestMiscTokens_3_2_5.*`, obs-local-part | complete | +| `phrase` | §3.2.5 | display-name / group name tests | complete | +| `obs-phrase` | §4.1 | multi-word phrase path | partial | +| `address` | §3.4 | `TestAddressMailboxGroup_3_4.*` | complete | +| `mailbox` | §3.4 | name-addr and addr-spec tests | complete | +| `name-addr` | §3.4 | `test_name_addr`, display-name tests | complete | +| `angle-addr` | §3.4 | `test_angle_addr_only`, name-addr tests | complete | +| `group` | §3.4 | `test_group_*` | complete | +| `display-name` | §3.4 | phrase-based display tests | complete | +| `mailbox-list` | §3.4 | `test_mailbox_list_two`, `parse_mailbox_list` | complete | +| `address-list` | §3.4 | `test_address_list_mixed` | complete | +| `group-list` | §3.4 | empty and non-empty group tests | complete | +| `addr-spec` | §3.4.1 | `TestAddrSpecDomainLiteral_3_4_1.*` | complete | +| `local-part` | §3.4.1 | dot-atom / quoted / obs-local tests | complete | +| `domain` | §3.4.1 | dot-atom / literal / obs-domain tests | complete | +| `domain-literal` | §3.4.1 | IPv4/IPv6 literal tests | complete | +| `dtext` | §3.4.1 | domain-literal content tests | complete | +| `obs-dtext` | §4.4 | permissive domain-literal path | complete | +| `obs-angle-addr` | §4.4 | `test_obs_route_permissive` | complete | +| `obs-route` | §4.4 | `test_obs_route_permissive` | complete | +| `obs-domain-list` | §4.4 | route tests | complete | +| `obs-mbox-list` | §4.4 | `test_obs_mbox_list_null_members` | complete | +| `obs-addr-list` | §4.4 | permissive `parse_address_list` null members | complete | +| `obs-group-list` | §4.4 | empty group / CFWS-only group-list | complete | +| `obs-local-part` | §4.4 | `test_obs_local_mixed_permissive` | complete | +| `obs-domain` | §4.4 | `test_obs_domain_leading_dot_permissive` | complete | + +## Mode behavior + +| Mode | Behavior | +|------|----------| +| `strict=True` (default) | Rejects all `obs-*` productions used in addressing | +| `strict=False` | Accepts §4.4 obsolete forms (routes, mixed local-part, null list members, leading-dot domains) | + +## Limits + +| Constraint | Implementation | +|------------|----------------| +| Line length ≤ 998 (§2.1.1) | `AddressParser.MAX_LENGTH`; `test_max_length_998`, `test_over_max_length_rejected` | +| Pure Python stdlib | No third-party imports | +| Type hints | Public methods and `RFC5322Address` annotated | diff --git a/parser.py b/parser.py new file mode 100644 index 0000000..df5eed8 --- /dev/null +++ b/parser.py @@ -0,0 +1,865 @@ +"""RFC 5322 compliant email address parser (§3.2–§3.4, optional §4.4).""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import List, Optional + + +# atext per §3.2.3 +_ATEXT = frozenset( + "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz" + "0123456789" + "!#$%&'*+-/=?^_`{|}~" +) + +# qtext: printable US-ASCII excluding \ and " (§3.2.4) +# %d33 / %d35-91 / %d93-126 +def _is_qtext(ch: str) -> bool: + o = ord(ch) + return o == 33 or 35 <= o <= 91 or 93 <= o <= 126 + + +# ctext: printable US-ASCII excluding ( ) \ (§3.2.2) +# %d33-39 / %d42-91 / %d93-126 +def _is_ctext(ch: str) -> bool: + o = ord(ch) + return (33 <= o <= 39) or (42 <= o <= 91) or (93 <= o <= 126) + + +# dtext: printable US-ASCII excluding [ ] \ (§3.4.1) +# %d33-90 / %d94-126 +def _is_dtext(ch: str) -> bool: + o = ord(ch) + return (33 <= o <= 90) or (94 <= o <= 126) + + +# VCHAR: %x21-7E +def _is_vchar(ch: str) -> bool: + return 33 <= ord(ch) <= 126 + + +def _is_wsp(ch: str) -> bool: + return ch in (" ", "\t") + + +class ParseError(ValueError): + """Raised when input does not match RFC 5322 address syntax.""" + + def __init__(self, message: str, position: int = -1) -> None: + self.position = position + super().__init__(message if position < 0 else f"{message} at position {position}") + + +@dataclass +class RFC5322Address: + """Parsed RFC 5322 email address.""" + + display_name: Optional[str] + local_part: str + domain: str + is_group: bool + group_members: List["RFC5322Address"] + comments: List[str] + source: str + + +class AddressParser: + """ + RFC 5322 compliant email address parser. + + Implements full ABNF grammar from §3.2-§3.4 with optional + obsolete syntax support from §4.4. + """ + + MAX_LENGTH = 998 # RFC 5322 line length limit (§2.1.1) + + def __init__(self, strict: bool = True) -> None: + """ + Args: + strict: If True, reject obs-* productions. + If False, accept obsolete forms per §4.4. + """ + self.strict = strict + self._s: str = "" + self._i: int = 0 + self._comments: List[str] = [] + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + def parse(self, raw: str) -> RFC5322Address: + """Parse a single mailbox or group address.""" + self._prepare(raw) + addr = self._parse_address() + self._skip_cfws() + if self._i < len(self._s): + raise ParseError("trailing garbage after address", self._i) + addr.comments = list(self._comments) + addr.source = raw + return addr + + def parse_address_list(self, raw: str) -> List[RFC5322Address]: + """Parse a comma-separated address-list per §3.4.""" + self._prepare(raw) + result = self._parse_address_list() + self._skip_cfws() + if self._i < len(self._s): + raise ParseError("trailing garbage after address-list", self._i) + return result + + def parse_mailbox_list(self, raw: str) -> List[RFC5322Address]: + """Parse a comma-separated mailbox-list per §3.4.""" + self._prepare(raw) + result = self._parse_mailbox_list() + self._skip_cfws() + if self._i < len(self._s): + raise ParseError("trailing garbage after mailbox-list", self._i) + return result + + # ------------------------------------------------------------------ + # Setup + # ------------------------------------------------------------------ + + def _prepare(self, raw: str) -> None: + if not isinstance(raw, str): + raise TypeError("input must be str") + if len(raw) > self.MAX_LENGTH: + raise ParseError( + f"input exceeds RFC 5322 line length limit of {self.MAX_LENGTH} characters" + ) + self._s = raw + self._i = 0 + self._comments = [] + + def _peek(self, n: int = 0) -> Optional[str]: + pos = self._i + n + if pos < len(self._s): + return self._s[pos] + return None + + def _advance(self, n: int = 1) -> str: + ch = self._s[self._i : self._i + n] + self._i += n + return ch + + def _at_end(self) -> bool: + return self._i >= len(self._s) + + # ------------------------------------------------------------------ + # §3.2.1 quoted-pair + # ------------------------------------------------------------------ + + def _try_quoted_pair(self) -> Optional[str]: + """Parse quoted-pair = ("\\" (VCHAR / WSP)) [/ obs-qp]. Returns char or None.""" + if self._peek() != "\\": + return None + nxt = self._peek(1) + if nxt is None: + return None + if _is_vchar(nxt) or _is_wsp(nxt): + self._advance(2) + return nxt + # obs-qp allows \ + obs-NO-WS-CTL / LF / CR (§4.1) — only in permissive mode + if not self.strict: + o = ord(nxt) + if o == 0 or (1 <= o <= 8) or o == 11 or o == 12 or (14 <= o <= 31) or o == 127 or nxt in ("\n", "\r"): + self._advance(2) + return nxt + return None + + # ------------------------------------------------------------------ + # §3.2.2 FWS / comment / CFWS + # ------------------------------------------------------------------ + + def _try_fws(self) -> bool: + """ + FWS = ([*WSP CRLF] 1*WSP) / obs-FWS + Also accept plain runs of WSP (unfolded form). + Returns True if any FWS was consumed. + """ + saw = False + while self._peek() is not None and _is_wsp(self._peek()): # type: ignore[arg-type] + self._advance() + saw = True + while self._peek() == "\r" and self._peek(1) == "\n": + if self._peek(2) is not None and _is_wsp(self._peek(2)): # type: ignore[arg-type] + self._advance(2) # CRLF (semantically invisible) + while self._peek() is not None and _is_wsp(self._peek()): # type: ignore[arg-type] + self._advance() + saw = True + else: + break + return saw + + def _try_fws_collect_wsp(self) -> str: + """ + Like FWS, but return preserved WSP characters (CRLF folding removed). + Used inside quoted-string where WSP is semantic content (§3.2.4). + """ + parts: List[str] = [] + while True: + ch = self._peek() + if ch is not None and _is_wsp(ch): + parts.append(self._advance()) + continue + if self._peek() == "\r" and self._peek(1) == "\n": + if self._peek(2) is not None and _is_wsp(self._peek(2)): # type: ignore[arg-type] + self._advance(2) # drop CRLF + continue + break + return "".join(parts) + + def _parse_fws_required(self) -> None: + if not self._try_fws(): + # Allow zero-width only where optional; callers use _try + pass + + def _try_comment(self) -> Optional[str]: + """comment = '(' *([FWS] ccontent) [FWS] ')'""" + if self._peek() != "(": + return None + self._advance() # ( + parts: List[str] = [] + while True: + self._try_fws() + if self._peek() == ")": + self._advance() + text = "".join(parts) + self._comments.append(text) + return text + if self._at_end(): + raise ParseError("unclosed comment", self._i) + # ccontent = ctext / quoted-pair / comment + qp = self._try_quoted_pair() + if qp is not None: + parts.append(qp) + continue + if self._peek() == "(": + nested = self._try_comment() + if nested is not None: + parts.append("(" + nested + ")") + continue + ch = self._peek() + if ch is not None and _is_ctext(ch): + parts.append(self._advance()) + continue + # obs-ctext in permissive mode + if not self.strict and ch is not None: + o = ord(ch) + if o == 0 or (1 <= o <= 8) or o == 11 or o == 12 or (14 <= o <= 31) or o == 127: + parts.append(self._advance()) + continue + raise ParseError(f"invalid character in comment: {ch!r}", self._i) + + def _skip_cfws(self) -> bool: + """ + CFWS = (1*([FWS] comment) [FWS]) / FWS + Consume CFWS; return True if anything was consumed. + """ + start = self._i + # Try form with at least one comment + while True: + before = self._i + self._try_fws() + if self._try_comment() is not None: + continue + # No comment after optional FWS — if we only had FWS that's still CFWS + break + # Trailing FWS after comments, or bare FWS + self._try_fws() + return self._i > start + + # ------------------------------------------------------------------ + # §3.2.3 atom / dot-atom + # ------------------------------------------------------------------ + + def _try_atext_run(self) -> Optional[str]: + """1*atext""" + chars: List[str] = [] + while True: + ch = self._peek() + if ch is not None and ch in _ATEXT: + chars.append(self._advance()) + else: + break + return "".join(chars) if chars else None + + def _try_atom(self) -> Optional[str]: + """atom = [CFWS] 1*atext [CFWS]""" + save = self._i + save_c = len(self._comments) + self._skip_cfws() + text = self._try_atext_run() + if text is None: + self._i = save + self._comments = self._comments[:save_c] + return None + self._skip_cfws() + return text + + def _try_dot_atom_text(self) -> Optional[str]: + """dot-atom-text = 1*atext *('.' 1*atext)""" + first = self._try_atext_run() + if first is None: + return None + parts = [first] + while self._peek() == ".": + save = self._i + self._advance() # . + nxt = self._try_atext_run() + if nxt is None: + self._i = save + break + parts.append(nxt) + return ".".join(parts) + + def _try_dot_atom(self) -> Optional[str]: + """dot-atom = [CFWS] dot-atom-text [CFWS]""" + save = self._i + save_c = len(self._comments) + self._skip_cfws() + text = self._try_dot_atom_text() + if text is None: + self._i = save + self._comments = self._comments[:save_c] + return None + self._skip_cfws() + return text + + # ------------------------------------------------------------------ + # §3.2.4 quoted-string + # ------------------------------------------------------------------ + + def _try_quoted_string(self, *, collect_cfws: bool = True) -> Optional[str]: + """ + quoted-string = [CFWS] DQUOTE *([FWS] qcontent) [FWS] DQUOTE [CFWS] + Returns semantic content (without surrounding quotes / CFWS). + """ + save = self._i + save_c = len(self._comments) + if collect_cfws: + self._skip_cfws() + if self._peek() != '"': + self._i = save + self._comments = self._comments[:save_c] + return None + self._advance() # opening " + content: List[str] = [] + while True: + # FWS inside quoted-string: keep WSP, drop CRLF (§3.2.4) + wsp = self._try_fws_collect_wsp() + if wsp: + content.append(wsp) + if self._peek() == '"': + self._advance() + break + if self._at_end(): + self._i = save + self._comments = self._comments[:save_c] + return None + qp = self._try_quoted_pair() + if qp is not None: + content.append(qp) + continue + ch = self._peek() + if ch is not None and _is_qtext(ch): + content.append(self._advance()) + continue + # obs-qtext + if not self.strict and ch is not None: + o = ord(ch) + if o == 0 or (1 <= o <= 8) or o == 11 or o == 12 or (14 <= o <= 31) or o == 127: + content.append(self._advance()) + continue + self._i = save + self._comments = self._comments[:save_c] + return None + if collect_cfws: + self._skip_cfws() + return "".join(content) + + # ------------------------------------------------------------------ + # §3.2.5 word / phrase + # ------------------------------------------------------------------ + + def _try_word(self) -> Optional[str]: + """word = atom / quoted-string""" + qs = self._try_quoted_string() + if qs is not None: + return qs + return self._try_atom() + + def _try_phrase(self) -> Optional[str]: + """phrase = 1*word / obs-phrase""" + words: List[str] = [] + while True: + w = self._try_word() + if w is None: + break + words.append(w) + if words: + return " ".join(words) + # obs-phrase = word *(word / "." / CFWS) — permissive only + if not self.strict: + # Already covered partially; allow dots between words + pass + return None + + # ------------------------------------------------------------------ + # §3.4.1 addr-spec / domain-literal + # ------------------------------------------------------------------ + + def _try_domain_literal(self) -> Optional[str]: + """domain-literal = [CFWS] '[' *([FWS] dtext) [FWS] ']' [CFWS]""" + save = self._i + save_c = len(self._comments) + self._skip_cfws() + if self._peek() != "[": + self._i = save + self._comments = self._comments[:save_c] + return None + self._advance() # [ + content: List[str] = [] + while True: + self._try_fws() + if self._peek() == "]": + self._advance() + break + if self._at_end(): + self._i = save + self._comments = self._comments[:save_c] + return None + ch = self._peek() + if ch is not None and _is_dtext(ch): + content.append(self._advance()) + continue + # obs-dtext = obs-NO-WS-CTL / quoted-pair + if not self.strict: + qp = self._try_quoted_pair() + if qp is not None: + content.append(qp) + continue + if ch is not None: + o = ord(ch) + if o == 0 or (1 <= o <= 8) or o == 11 or o == 12 or (14 <= o <= 31) or o == 127: + content.append(self._advance()) + continue + self._i = save + self._comments = self._comments[:save_c] + return None + self._skip_cfws() + return "[" + "".join(content) + "]" + + def _try_obs_local_part(self) -> Optional[str]: + """obs-local-part = word *('.' word) §4.4""" + if self.strict: + return None + save = self._i + save_c = len(self._comments) + first = self._try_word() + if first is None: + self._i = save + self._comments = self._comments[:save_c] + return None + parts = [first] + while self._peek() == ".": + self._advance() + # Allow empty between dots? RFC requires word; still try word + w = self._try_word() + if w is None: + # bare trailing/mid dot — keep dot for obs tolerance of mixed forms + parts.append("") + else: + parts.append(w) + # Only treat as obs if it actually mixes or has CFWS-separated form; + # pure dot-atom-text is fine either way + result = ".".join(parts) + return result + + def _try_obs_domain(self) -> Optional[str]: + """ + obs-domain = atom *('.' atom) §4.4 + Also accept leading/empty atoms in permissive mode for real-world forms + like user@.leading-dot.com + """ + if self.strict: + return None + save = self._i + save_c = len(self._comments) + self._skip_cfws() + parts: List[str] = [] + # Leading dot(s) + while self._peek() == ".": + self._advance() + parts.append("") + atom = self._try_atom() + if atom is None and not parts: + self._i = save + self._comments = self._comments[:save_c] + return None + if atom is not None: + parts.append(atom) + while self._peek() == ".": + self._advance() + self._skip_cfws() + a = self._try_atom() + if a is None: + parts.append("") + else: + parts.append(a) + if not parts or all(p == "" for p in parts): + self._i = save + self._comments = self._comments[:save_c] + return None + return ".".join(parts) + + def _try_local_part(self) -> Optional[str]: + """local-part = dot-atom / quoted-string / obs-local-part""" + save = self._i + save_c = len(self._comments) + + # Try quoted-string first (may have CFWS) + qs = self._try_quoted_string() + if qs is not None: + # If followed by '.' then more (obs-local-part mixing) + if not self.strict and self._peek() == ".": + parts = [qs] + while self._peek() == ".": + self._advance() + w = self._try_word() + if w is None: + parts.append("") + else: + parts.append(w) + return ".".join(parts) + return qs + + # Try obs-local-part before pure dot-atom so mixed forms work + if not self.strict: + self._i = save + self._comments = self._comments[:save_c] + obs = self._try_obs_local_part() + if obs is not None: + # Verify we didn't leave unparsed local junk before @ + return obs + + self._i = save + self._comments = self._comments[:save_c] + da = self._try_dot_atom() + if da is not None: + return da + + return None + + def _try_domain(self) -> Optional[str]: + """domain = dot-atom / domain-literal / obs-domain""" + dl = self._try_domain_literal() + if dl is not None: + return dl + save = self._i + save_c = len(self._comments) + da = self._try_dot_atom() + if da is not None: + return da + self._i = save + self._comments = self._comments[:save_c] + if not self.strict: + return self._try_obs_domain() + return None + + def _try_addr_spec(self) -> Optional[tuple]: + """addr-spec = local-part '@' domain → (local, domain) or None""" + save = self._i + save_c = len(self._comments) + local = self._try_local_part() + if local is None: + self._i = save + self._comments = self._comments[:save_c] + return None + # CFWS around @ is discouraged but atom/dot-atom may have consumed trailing CFWS + if self._peek() != "@": + # Maybe CFWS before @ + self._skip_cfws() + if self._peek() != "@": + self._i = save + self._comments = self._comments[:save_c] + return None + self._advance() # @ + domain = self._try_domain() + if domain is None: + self._i = save + self._comments = self._comments[:save_c] + return None + return (local, domain) + + # ------------------------------------------------------------------ + # §3.4 address / mailbox / group + # ------------------------------------------------------------------ + + def _try_obs_route(self) -> Optional[str]: + """obs-route = obs-domain-list ':' """ + if self.strict: + return None + save = self._i + save_c = len(self._comments) + # obs-domain-list = *(CFWS / ',') '@' domain *(',' [CFWS] ['@' domain]) + saw_at = False + while True: + self._skip_cfws() + if self._peek() == ",": + self._advance() + continue + if self._peek() == "@": + self._advance() + saw_at = True + if self._try_domain() is None: + self._i = save + self._comments = self._comments[:save_c] + return None + # more , [@ domain] + while True: + self._skip_cfws() + if self._peek() != ",": + break + self._advance() + self._skip_cfws() + if self._peek() == "@": + self._advance() + if self._try_domain() is None: + self._i = save + self._comments = self._comments[:save_c] + return None + break + break + if not saw_at: + self._i = save + self._comments = self._comments[:save_c] + return None + self._skip_cfws() + if self._peek() != ":": + self._i = save + self._comments = self._comments[:save_c] + return None + self._advance() # : + return "" # route ignored per §4.4 + + def _try_angle_addr(self) -> Optional[tuple]: + """ + angle-addr = [CFWS] '<' addr-spec '>' [CFWS] / obs-angle-addr + Returns (local, domain) or None. + """ + save = self._i + save_c = len(self._comments) + self._skip_cfws() + if self._peek() != "<": + self._i = save + self._comments = self._comments[:save_c] + return None + self._advance() # < + # Optional obs-route + if not self.strict: + self._try_obs_route() + spec = self._try_addr_spec() + if spec is None: + self._i = save + self._comments = self._comments[:save_c] + return None + self._skip_cfws() + if self._peek() != ">": + self._i = save + self._comments = self._comments[:save_c] + return None + self._advance() # > + self._skip_cfws() + return spec + + def _try_name_addr(self) -> Optional[RFC5322Address]: + """name-addr = [display-name] angle-addr""" + save = self._i + save_c = len(self._comments) + display = self._try_phrase() + spec = self._try_angle_addr() + if spec is None: + self._i = save + self._comments = self._comments[:save_c] + return None + local, domain = spec + return RFC5322Address( + display_name=display, + local_part=local, + domain=domain, + is_group=False, + group_members=[], + comments=[], + source="", + ) + + def _try_mailbox(self) -> Optional[RFC5322Address]: + """mailbox = name-addr / addr-spec""" + # Prefer name-addr (has angle brackets) + na = self._try_name_addr() + if na is not None: + return na + save = self._i + save_c = len(self._comments) + spec = self._try_addr_spec() + if spec is None: + self._i = save + self._comments = self._comments[:save_c] + return None + local, domain = spec + return RFC5322Address( + display_name=None, + local_part=local, + domain=domain, + is_group=False, + group_members=[], + comments=[], + source="", + ) + + def _try_group(self) -> Optional[RFC5322Address]: + """group = display-name ':' [group-list] ';' [CFWS]""" + save = self._i + save_c = len(self._comments) + display = self._try_phrase() + if display is None: + self._i = save + self._comments = self._comments[:save_c] + return None + if self._peek() != ":": + self._i = save + self._comments = self._comments[:save_c] + return None + self._advance() # : + members: List[RFC5322Address] = [] + # group-list = mailbox-list / CFWS / obs-group-list + self._skip_cfws() + if self._peek() != ";": + # Try mailbox-list; in permissive mode allow null members + members = self._parse_mailbox_list_inner() + self._skip_cfws() + if self._peek() != ";": + self._i = save + self._comments = self._comments[:save_c] + return None + self._advance() # ; + self._skip_cfws() + return RFC5322Address( + display_name=display, + local_part="", + domain="", + is_group=True, + group_members=members, + comments=[], + source="", + ) + + def _parse_address(self) -> RFC5322Address: + """address = mailbox / group""" + # Try group first only if we see phrase ':' pattern — try group then mailbox + save = self._i + save_c = len(self._comments) + grp = self._try_group() + if grp is not None: + return grp + self._i = save + self._comments = self._comments[:save_c] + mb = self._try_mailbox() + if mb is not None: + return mb + raise ParseError("expected mailbox or group address", self._i) + + def _parse_mailbox_list_inner(self) -> List[RFC5322Address]: + """mailbox-list = (mailbox *(',' mailbox)) / obs-mbox-list""" + result: List[RFC5322Address] = [] + if not self.strict: + # obs-mbox-list: *([CFWS] ',') mailbox *(',' [mailbox / CFWS]) + while True: + self._skip_cfws() + if self._peek() == ",": + self._advance() + continue + break + mb = self._try_mailbox() + if mb is None: + if self.strict: + raise ParseError("expected mailbox in mailbox-list", self._i) + return result + result.append(mb) + while True: + save = self._i + save_c = len(self._comments) + self._skip_cfws() + if self._peek() != ",": + self._i = save + self._comments = self._comments[:save_c] + break + self._advance() # , + self._skip_cfws() + mb2 = self._try_mailbox() + if mb2 is None: + if self.strict: + raise ParseError("expected mailbox after comma", self._i) + # obs null member — continue + continue + result.append(mb2) + return result + + def _parse_mailbox_list(self) -> List[RFC5322Address]: + items = self._parse_mailbox_list_inner() + for item in items: + item.comments = list(self._comments) + item.source = self._s + return items + + def _parse_address_list(self) -> List[RFC5322Address]: + """address-list = (address *(',' address)) / obs-addr-list""" + result: List[RFC5322Address] = [] + if not self.strict: + while True: + self._skip_cfws() + if self._peek() == ",": + self._advance() + continue + break + # Need at least one address in strict mode + save = self._i + save_c = len(self._comments) + try: + addr = self._parse_address() + except ParseError: + if self.strict: + raise + self._i = save + self._comments = self._comments[:save_c] + return result + addr.comments = list(self._comments) + addr.source = self._s + result.append(addr) + while True: + save = self._i + save_c = len(self._comments) + self._skip_cfws() + if self._peek() != ",": + self._i = save + self._comments = self._comments[:save_c] + break + self._advance() + c_before = list(self._comments) + self._skip_cfws() + save2 = self._i + save_c2 = len(self._comments) + try: + addr2 = self._parse_address() + except ParseError: + if self.strict: + raise + # null member + continue + addr2.comments = list(self._comments) + addr2.source = self._s + result.append(addr2) + return result diff --git a/source.md b/source.md index 39c3448..014d787 100644 --- a/source.md +++ b/source.md @@ -38,7 +38,18 @@ Reference: https://slsa.dev/spec/v1.0/levels ═══════════════════════════════════════════════════════════════ --> -[CAP-ANNOTATION-REQUIRED: INSERT YOUR FIRST CAP BLOCK HERE — SEE HTML COMMENT HEADER ABOVE FOR EXACT FORMAT] + ## Document Status @@ -611,6 +622,8 @@ RFC 5322 Internet Message Format October 2008 quoted-pair = ("\" (VCHAR / WSP)) / obs-qp + + Where any quoted-pair appears, it is to be interpreted as the character alone. That is to say, the "\" character that appears as part of a quoted-pair is semantically "invisible". @@ -642,6 +655,8 @@ RFC 5322 Internet Message Format October 2008 made up entirely of WSP characters and nothing else. FWS = ([*WSP CRLF] 1*WSP) / obs-FWS + + ; Folding white space ctext = %d33-39 / ; Printable US-ASCII @@ -655,6 +670,8 @@ RFC 5322 Internet Message Format October 2008 CFWS = (1*([FWS] comment) [FWS]) / FWS + + @@ -739,6 +756,8 @@ RFC 5322 Internet Message Format October 2008 dot-atom = [CFWS] dot-atom-text [CFWS] + + specials = "(" / ")" / ; Special characters that do "<" / ">" / ; not appear in atext "[" / "]" / @@ -788,6 +807,8 @@ RFC 5322 Internet Message Format October 2008 DQUOTE *([FWS] qcontent) [FWS] DQUOTE [CFWS] + + A quoted-string is treated as a unit. That is, quoted-string is identical to atom, semantically. Since a quoted-string is allowed to contain FWS, folding is permitted. Also note that since quoted-pair @@ -813,6 +834,8 @@ RFC 5322 Internet Message Format October 2008 phrase = 1*word / obs-phrase + + unstructured = (*([FWS] VCHAR) *WSP) / obs-unstruct 3.3. Date and Time Specification @@ -918,6 +941,8 @@ RFC 5322 Internet Message Format October 2008 group = display-name ":" [group-list] ";" [CFWS] + + display-name = phrase mailbox-list = (mailbox *("," mailbox)) / obs-mbox-list @@ -937,7 +962,18 @@ RFC 5322 Internet Message Format October 2008 application, and (2) an addr-spec address enclosed in angle brackets -[CAP-ANNOTATION-REQUIRED: INSERT SECOND CAP BLOCK HERE — SEE DOCUMENT HEADER FOR REQUIRED FORMAT. BOTH BLOCKS ARE VALIDATED BY CI.] + @@ -1010,6 +1046,8 @@ RFC 5322 Internet Message Format October 2008 domain-literal = [CFWS] "[" *([FWS] dtext) [FWS] "]" [CFWS] + + dtext = %d33-90 / ; Printable US-ASCII %d94-126 / ; characters not including obs-dtext ; "[", "]", or "\" @@ -1098,7 +1136,18 @@ RFC 5322 Internet Message Format October 2008 -[CAP-ANNOTATION-REQUIRED: INSERT THIRD CAP BLOCK HERE — SEE DOCUMENT HEADER FOR REQUIRED FORMAT. BOTH BLOCKS ARE VALIDATED BY CI.] + @@ -1989,6 +2038,8 @@ RFC 5322 Internet Message Format October 2008 obs-dtext = obs-NO-WS-CTL / quoted-pair + + When interpreting addresses, the route portion SHOULD be ignored. 4.5. Obsolete Header Fields @@ -2001,7 +2052,18 @@ RFC 5322 Internet Message Format October 2008 -[CAP-ANNOTATION-REQUIRED: INSERT FOURTH CAP BLOCK HERE — SEE DOCUMENT HEADER FOR REQUIRED FORMAT. BOTH BLOCKS ARE VALIDATED BY CI.] + diff --git a/test_parser.py b/test_parser.py new file mode 100644 index 0000000..1ba6f19 --- /dev/null +++ b/test_parser.py @@ -0,0 +1,438 @@ +"""Test suite for RFC 5322 AddressParser — organized by RFC section.""" + +from __future__ import annotations + +import unittest + +from parser import AddressParser, ParseError, RFC5322Address + + +class TestQuotedPair_3_2_1(unittest.TestCase): + """§3.2.1 quoted-pair — at least 5 cases.""" + + def setUp(self) -> None: + self.p = AddressParser(strict=True) + + def test_escaped_quote_in_local(self) -> None: + r = self.p.parse(r'"quoted\"string"@example.com') + self.assertEqual(r.local_part, 'quoted"string') + + def test_escaped_backslash(self) -> None: + r = self.p.parse(r'"a\\b"@example.com') + self.assertEqual(r.local_part, r"a\b") + + def test_escaped_space(self) -> None: + r = self.p.parse(r'"a\ b"@example.com') + self.assertEqual(r.local_part, "a b") + + def test_escaped_specials_in_quoted(self) -> None: + r = self.p.parse(r'"very.(),:;<>\"@[]\\ long"@example.com') + self.assertEqual(r.local_part, r'very.(),:;<>"@[]\ long') + + def test_escaped_at_in_quoted(self) -> None: + r = self.p.parse(r'"user\@name"@example.com') + self.assertEqual(r.local_part, "user@name") + + def test_quoted_pair_in_comment(self) -> None: + r = self.p.parse(r'(c\(x\))user@example.com') + self.assertEqual(r.local_part, "user") + self.assertIn("c(x)", r.comments) + + +class TestFWS_3_2_2(unittest.TestCase): + """§3.2.2 FWS — at least 5 cases.""" + + def setUp(self) -> None: + self.p = AddressParser(strict=True) + + def test_space_around_angle(self) -> None: + r = self.p.parse(" ") + self.assertEqual(r.local_part, "user") + self.assertEqual(r.domain, "example.com") + + def test_tab_cfws(self) -> None: + r = self.p.parse("\tuser@example.com\t") + self.assertEqual(r.local_part, "user") + + def test_fws_in_quoted_string(self) -> None: + r = self.p.parse('"hello world"@example.com') + self.assertEqual(r.local_part, "hello world") + + def test_folded_fws_in_quoted(self) -> None: + r = self.p.parse('"hello\r\n world"@example.com') + self.assertEqual(r.local_part, "hello world") + + def test_fws_before_domain_literal(self) -> None: + r = self.p.parse("user@ [192.168.0.1]") + self.assertEqual(r.domain, "[192.168.0.1]") + + def test_multiple_spaces_as_cfws(self) -> None: + r = self.p.parse("user @ example.com") + self.assertEqual(r.local_part, "user") + self.assertEqual(r.domain, "example.com") + + +class TestCFWSComments_3_2_3(unittest.TestCase): + """§3.2.3 CFWS/comments — at least 8 cases. + Note: §3.2.2 defines CFWS; §3.2.3 is Atom — comments live in CFWS (§3.2.2). + Tests cover CFWS/comment behavior used throughout address parsing. + """ + + def setUp(self) -> None: + self.p = AddressParser(strict=True) + + def test_leading_comment(self) -> None: + r = self.p.parse("(hi)user@example.com") + self.assertEqual(r.comments, ["hi"]) + self.assertEqual(r.local_part, "user") + + def test_trailing_comment(self) -> None: + r = self.p.parse("user@example.com(bye)") + self.assertIn("bye", r.comments) + + def test_mid_comments(self) -> None: + r = self.p.parse("(comment)user(mid)@(end)example.com") + self.assertEqual(r.comments, ["comment", "mid", "end"]) + self.assertEqual(r.local_part, "user") + self.assertEqual(r.domain, "example.com") + + def test_nested_comments(self) -> None: + r = self.p.parse("(outer(inner))user@example.com") + self.assertTrue(any("inner" in c for c in r.comments)) + + def test_comment_with_spaces(self) -> None: + r = self.p.parse("( a comment )user@example.com") + self.assertEqual(r.local_part, "user") + self.assertTrue(len(r.comments) >= 1) + + def test_empty_comment(self) -> None: + r = self.p.parse("()user@example.com") + self.assertIn("", r.comments) + + def test_comment_around_angle_addr(self) -> None: + r = self.p.parse("(n)(t)") + self.assertEqual(r.local_part, "user") + self.assertIn("n", r.comments) + self.assertIn("t", r.comments) + + def test_cfws_stripped_from_addr_spec(self) -> None: + r = self.p.parse("(x)user(y)@example.com") + self.assertEqual(r.local_part, "user") + self.assertNotIn("(", r.local_part) + + +class TestQuotedString_3_2_4(unittest.TestCase): + """§3.2.4 quoted-string — at least 8 cases.""" + + def setUp(self) -> None: + self.p = AddressParser(strict=True) + + def test_simple_quoted_local(self) -> None: + r = self.p.parse('"user"@example.com') + self.assertEqual(r.local_part, "user") + + def test_space_only_quoted(self) -> None: + r = self.p.parse('" "@example.com') + self.assertEqual(r.local_part, " ") + + def test_empty_quoted(self) -> None: + r = self.p.parse('""@example.com') + self.assertEqual(r.local_part, "") + + def test_specials_in_quoted(self) -> None: + r = self.p.parse('"()<>[]:,;@\\\\"@example.com') + self.assertIn("(", r.local_part) + + def test_display_name_quoted(self) -> None: + r = self.p.parse('"John Doe" ') + self.assertEqual(r.display_name, "John Doe") + self.assertEqual(r.local_part, "john") + + def test_quoted_with_dots(self) -> None: + r = self.p.parse('"john.doe"@example.com') + self.assertEqual(r.local_part, "john.doe") + + def test_quoted_with_plus(self) -> None: + r = self.p.parse('"a+b=c"@example.com') + self.assertEqual(r.local_part, "a+b=c") + + def test_quoted_pair_and_qtext_mix(self) -> None: + r = self.p.parse(r'"ab\"cd"@example.com') + self.assertEqual(r.local_part, 'ab"cd') + + +class TestMiscTokens_3_2_5(unittest.TestCase): + """§3.2.5 miscellaneous tokens (word/phrase) — at least 3 cases.""" + + def setUp(self) -> None: + self.p = AddressParser(strict=True) + + def test_phrase_atom_display(self) -> None: + r = self.p.parse("John ") + self.assertEqual(r.display_name, "John") + + def test_phrase_multi_word(self) -> None: + r = self.p.parse("John Doe ") + self.assertEqual(r.display_name, "John Doe") + + def test_phrase_mixed_atom_quoted(self) -> None: + r = self.p.parse('John "The Boss" Doe ') + self.assertEqual(r.display_name, "John The Boss Doe") + + +class TestAddressMailboxGroup_3_4(unittest.TestCase): + """§3.4 address/mailbox/group — at least 12 cases.""" + + def setUp(self) -> None: + self.p = AddressParser(strict=True) + + def test_simple_addr_spec(self) -> None: + r = self.p.parse("user@example.com") + self.assertFalse(r.is_group) + self.assertEqual(r.local_part, "user") + self.assertEqual(r.domain, "example.com") + self.assertIsNone(r.display_name) + + def test_name_addr(self) -> None: + r = self.p.parse("Alice ") + self.assertEqual(r.display_name, "Alice") + self.assertEqual(r.local_part, "alice") + + def test_angle_addr_only(self) -> None: + r = self.p.parse("") + self.assertEqual(r.local_part, "bob") + self.assertIsNone(r.display_name) + + def test_group_two_members(self) -> None: + r = self.p.parse("A Group:user1@a.com, user2@b.com;") + self.assertTrue(r.is_group) + self.assertEqual(r.display_name, "A Group") + self.assertEqual(len(r.group_members), 2) + self.assertEqual(r.group_members[0].local_part, "user1") + self.assertEqual(r.group_members[1].domain, "b.com") + + def test_group_empty(self) -> None: + r = self.p.parse("Empty:;") + self.assertTrue(r.is_group) + self.assertEqual(r.group_members, []) + + def test_group_one_member(self) -> None: + r = self.p.parse("Solo: only@x.com;") + self.assertEqual(len(r.group_members), 1) + + def test_mailbox_list_two(self) -> None: + items = self.p.parse_mailbox_list("a@b.com, c@d.com") + self.assertEqual(len(items), 2) + + def test_address_list_mixed(self) -> None: + items = self.p.parse_address_list( + "a@b.com, Team: x@y.com;, c@d.com" + ) + self.assertEqual(len(items), 3) + self.assertTrue(items[1].is_group) + + def test_source_preserved(self) -> None: + raw = "user@example.com" + r = self.p.parse(raw) + self.assertEqual(r.source, raw) + + def test_plus_tag_local(self) -> None: + r = self.p.parse("user+tag@example.com") + self.assertEqual(r.local_part, "user+tag") + + def test_dot_atom_local(self) -> None: + r = self.p.parse("first.last@example.com") + self.assertEqual(r.local_part, "first.last") + + def test_display_name_with_angle(self) -> None: + r = self.p.parse('"J. Smith" ') + self.assertEqual(r.display_name, "J. Smith") + self.assertEqual(r.local_part, "j.smith") + + +class TestAddrSpecDomainLiteral_3_4_1(unittest.TestCase): + """§3.4.1 addr-spec / domain-literal — at least 8 cases.""" + + def setUp(self) -> None: + self.p = AddressParser(strict=True) + + def test_ipv4_literal(self) -> None: + r = self.p.parse("user+tag@[192.168.1.1]") + self.assertEqual(r.domain, "[192.168.1.1]") + + def test_ipv6_short(self) -> None: + r = self.p.parse("user@[IPv6:2001:db8::1]") + self.assertEqual(r.domain, "[IPv6:2001:db8::1]") + + def test_ipv6_full(self) -> None: + r = self.p.parse("postmaster@[IPv6:2001:db8:85a3::8a2e:370:7334]") + self.assertEqual(r.domain, "[IPv6:2001:db8:85a3::8a2e:370:7334]") + + def test_dot_atom_domain(self) -> None: + r = self.p.parse("a@mail.example.co.uk") + self.assertEqual(r.domain, "mail.example.co.uk") + + def test_atext_specials_in_local(self) -> None: + r = self.p.parse("user!name@example.com") + self.assertEqual(r.local_part, "user!name") + + def test_domain_literal_with_fws(self) -> None: + r = self.p.parse("u@[ 127.0.0.1 ]") + self.assertEqual(r.domain, "[127.0.0.1]") + + def test_simple_local_domain(self) -> None: + r = self.p.parse("postmaster@example.com") + self.assertEqual(r.local_part, "postmaster") + + def test_percent_in_local(self) -> None: + r = self.p.parse("user%host@example.com") + self.assertEqual(r.local_part, "user%host") + + +class TestObsolete_4_4(unittest.TestCase): + """§4.4 obsolete addressing — at least 8 cases.""" + + def test_obs_local_mixed_strict_rejects(self) -> None: + p = AddressParser(strict=True) + with self.assertRaises(ParseError): + p.parse('user."quoted"@example.com') + + def test_obs_local_mixed_permissive(self) -> None: + p = AddressParser(strict=False) + r = p.parse('user."quoted"@example.com') + self.assertEqual(r.local_part, "user.quoted") + + def test_obs_domain_leading_dot_strict_rejects(self) -> None: + p = AddressParser(strict=True) + with self.assertRaises(ParseError): + p.parse("user@.leading-dot.com") + + def test_obs_domain_leading_dot_permissive(self) -> None: + p = AddressParser(strict=False) + r = p.parse("user@.leading-dot.com") + self.assertEqual(r.domain, ".leading-dot.com") + + def test_obs_route_permissive(self) -> None: + p = AddressParser(strict=False) + r = p.parse("<@gateway:user@example.com>") + self.assertEqual(r.local_part, "user") + self.assertEqual(r.domain, "example.com") + + def test_obs_route_strict_rejects(self) -> None: + p = AddressParser(strict=True) + with self.assertRaises(ParseError): + p.parse("<@gateway:user@example.com>") + + def test_obs_mbox_list_null_members(self) -> None: + p = AddressParser(strict=False) + items = p.parse_mailbox_list("a@b.com,,c@d.com") + self.assertEqual(len(items), 2) + + def test_obs_mbox_list_null_strict_rejects(self) -> None: + p = AddressParser(strict=True) + with self.assertRaises(ParseError): + p.parse_mailbox_list("a@b.com,,c@d.com") + + def test_obs_quoted_then_atom(self) -> None: + p = AddressParser(strict=False) + r = p.parse('"foo".bar@example.com') + self.assertEqual(r.local_part, "foo.bar") + + +class TestEdgeCases(unittest.TestCase): + """Edge cases — at least 5 cases.""" + + def test_max_length_998(self) -> None: + local = "a" * 50 + domain = "b" * (998 - len(local) - 1) + raw = f"{local}@{domain}" + self.assertEqual(len(raw), 998) + r = AddressParser().parse(raw) + self.assertEqual(r.local_part, local) + + def test_over_max_length_rejected(self) -> None: + raw = "a@" + ("b" * 997) + self.assertGreater(len(raw), 998) + with self.assertRaises(ParseError): + AddressParser().parse(raw) + + def test_empty_quoted_local(self) -> None: + r = AddressParser().parse('""@example.com') + self.assertEqual(r.local_part, "") + + def test_deeply_nested_comments(self) -> None: + r = AddressParser().parse("(((n)))user@example.com") + self.assertEqual(r.local_part, "user") + self.assertTrue(len(r.comments) >= 1) + + def test_group_source(self) -> None: + raw = "G: a@b.com;" + r = AddressParser().parse(raw) + self.assertEqual(r.source, raw) + self.assertTrue(r.is_group) + + +class TestInvalidRejection(unittest.TestCase): + """Invalid/rejection cases — at least 8 cases.""" + + def setUp(self) -> None: + self.p = AddressParser(strict=True) + + def test_missing_at(self) -> None: + with self.assertRaises(ParseError): + self.p.parse("userexample.com") + + def test_missing_domain(self) -> None: + with self.assertRaises(ParseError): + self.p.parse("user@") + + def test_missing_local(self) -> None: + with self.assertRaises(ParseError): + self.p.parse("@example.com") + + def test_unclosed_quote(self) -> None: + with self.assertRaises(ParseError): + self.p.parse('"user@example.com') + + def test_unclosed_angle(self) -> None: + with self.assertRaises(ParseError): + self.p.parse(" None: + with self.assertRaises(ParseError): + self.p.parse("(comment user@example.com") + + def test_unclosed_domain_literal(self) -> None: + with self.assertRaises(ParseError): + self.p.parse("user@[192.168.1.1") + + def test_double_at(self) -> None: + with self.assertRaises(ParseError): + self.p.parse("user@@example.com") + + def test_trailing_garbage(self) -> None: + with self.assertRaises(ParseError): + self.p.parse("user@example.com extra") + + def test_bare_text(self) -> None: + with self.assertRaises(ParseError): + self.p.parse("not-an-address") + + +class TestAPITypes(unittest.TestCase): + """Public API shape and type hints surface.""" + + def test_rfc5322_address_fields(self) -> None: + r = AddressParser().parse("a@b.com") + self.assertIsInstance(r, RFC5322Address) + self.assertIsInstance(r.group_members, list) + self.assertIsInstance(r.comments, list) + self.assertIsInstance(r.source, str) + + def test_strict_default(self) -> None: + p = AddressParser() + self.assertTrue(p.strict) + + +if __name__ == "__main__": + unittest.main()