Add support for parsing CSeq headers

src/sipmessage/__init__.py

@@ -6,9 +6,10 @@
 import importlib.metadata
 
 from .address import Address
+from .cseq import CSeq
 from .parameters import Parameters
 from .uri import URI
 from .via import Via
 
-__all__ = ["Address", "Parameters", "URI", "Via"]
+__all__ = ["Address", "CSeq", "Parameters", "URI", "Via"]
 __version__ = importlib.metadata.version("sipmessage")

src/sipmessage/address.py

@@ -18,8 +18,7 @@
         # display-name
         f"^(?P<name>{TOKEN_LWS}*|{grammar.QUOTED_STRING})"
         # LAQUOT addr-spec RAQUOT
-        "[ ]*"
-        "<(?P<uri>[^>]+)>"
+        f"{grammar.SWS}<(?P<uri>[^>]+)>{grammar.SWS}"
         # *(SEMI contact-params)
         f"(?P<parameters>(?:{grammar.SEMI}{grammar.GENERIC_PARAM})*)"
         "$"

src/sipmessage/cseq.py

@@ -0,0 +1,45 @@
+#
+# Copyright (C) Spacinov SAS
+# Distributed under the 2-clause BSD license
+#
+
+import dataclasses
+import re
+
+from . import grammar
+
+CSEQ_PATTERN = re.compile(
+    f"^(?P<sequence>{grammar.DIGIT}+){grammar.LWS}(?P<method>{grammar.TOKEN})$"
+)
+
+
+@dataclasses.dataclass
+class CSeq:
+    """
+    A `CSeq` header, used to identity and order transactions.
+    """
+
+    sequence: int
+    "The sequence number."
+
+    method: str
+    "The request method."
+
+    @classmethod
+    def parse(cls, value: str) -> "CSeq":
+        """
+        Parse the given string into an :class:`Address` instance.
+
+        If parsing fails, a :class:`ValueError` is raised.
+        """
+        pass
+        value = grammar.simplify_whitespace(value)
+
+        m = CSEQ_PATTERN.match(value)
+        if m is None:
+            raise ValueError("CSeq is not valid")
+
+        return cls(sequence=int(m.group("sequence")), method=m.group("method"))
+
+    def __str__(self) -> str:
+        return f"{self.sequence} {self.method}"

src/sipmessage/grammar.py

@@ -38,18 +38,21 @@ def simplify_whitespace(value: str) -> str:
 # Regular expression fragments.
 ALPHA = cset(C_ALPHA)
 ALPHANUM = cset(C_ALPHANUM)
+DIGIT = "\\d"
 HEXDIG = cset(string.hexdigits)
 ESCAPED = f"%{HEXDIG}{{2}}"
+LWS = "[ ]+"
+SWS = "[ ]*"
 TOKEN = f"{cset(C_TOKEN)}+"
 QUOTED_STRING = '"(?:[^"]|\\")*"'
 
-EQUAL = "[ ]*=[ ]*"
-SEMI = "[ ]*;[ ]*"
-SLASH = "[ ]*/[ ]*"
+EQUAL = f"{SWS}={SWS}"
+SEMI = f"{SWS};{SWS}"
+SLASH = f"{SWS}/{SWS}"
 
 HEXSEQ = f"{HEXDIG}{{1,4}}(?::{HEXDIG}{{1,4}})*"
 HEXPART = f"(?:{HEXSEQ}|{HEXSEQ}::(?:{HEXSEQ})?|::(?:{HEXSEQ})?)"
-IPV4ADDRESS = "\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}"
+IPV4ADDRESS = f"{DIGIT}{{1,3}}\\.{DIGIT}{{1,3}}\\.{DIGIT}{{1,3}}\\.{DIGIT}{{1,3}}"
 IPV6ADDRESS = f"{HEXPART}(?::{IPV4ADDRESS})?"
 IPV6REFERENCE = f"\\[{IPV6ADDRESS}\\]"
 
@@ -60,7 +63,7 @@ def simplify_whitespace(value: str) -> str:
 HOST = f"(?:{HOSTNAME}|{IPV4ADDRESS}|{IPV6REFERENCE})"
 USER = f"(?:{cset(C_USER_SAFE)}|{ESCAPED})+"
 PASSWORD = f"(?:{cset(C_PASSWORD_SAFE)}|{ESCAPED})+"
-PORT = "\\d+"
+PORT = f"{DIGIT}+"
 
 URI_PARAMCHAR = f"(?:{cset(C_URI_PARAM_SAFE)}|{ESCAPED})"
 URI_PARAM = f"{URI_PARAMCHAR}+(?:={URI_PARAMCHAR}+)?"

tests/test_cseq.py

@@ -0,0 +1,20 @@
+#
+# Copyright (C) Spacinov SAS
+# Distributed under the 2-clause BSD license
+#
+
+import unittest
+
+from sipmessage import CSeq
+
+
+class ViaTest(unittest.TestCase):
+    def test_empty(self) -> None:
+        with self.assertRaises(ValueError) as cm:
+            CSeq.parse("")
+        self.assertEqual(str(cm.exception), "CSeq is not valid")
+
+    def test_rfc4475_wsinv(self) -> None:
+        cseq = CSeq.parse("  0009\r\n  INVITE ")
+        self.assertEqual(cseq, CSeq(sequence=9, method="INVITE"))
+        self.assertEqual(str(cseq), "9 INVITE")