martians: overhaul

- Use the XML files instead of the CSV
- Separate output in inbound and outbound tables
- Add script for producing all the outputs

utils/openbsd/pf-martians/iana-ip-special-registry-parser.py

@@ -1,38 +1,101 @@
 #!/usr/bin/env python3
-import csv
-import re
+import argparse
+from enum import Enum
 import sys
+import xml.sax
 
-REGEX_FOOTNOTES = re.compile(r" \[\d+\]")
-REGEX_WHITESPACES = re.compile(r"\s+")
 
-def parse_iana_csv(csvfile):
-	reader = csv.DictReader(csvfile)
-	for row in reader:
-		fmt = ""
-		reachable = row["Globally Reachable"]
-		if reachable == "":
-			continue
-		if "False" in reachable:
-			fmt = "{}"
-		elif "True" in reachable:
-			fmt = "!{}"
-		elif "N/A" in reachable:
-			fmt = "#{}"
-		else:
-			fmt = "#[" + reachable + "]: {}"
-		blocks = row["Address Block"]
-		blocks = REGEX_FOOTNOTES.sub("", blocks)
-		blocks = REGEX_WHITESPACES.sub("", blocks)
-		for block in blocks.split(","):
-			print(fmt.format(block))
+class ParserState(Enum):
+	NOOP = 0
+	DISCARD = 1
+	CAPTURE = 2
+
+
+class IPDisallowedHandler(xml.sax.ContentHandler):
+	TRANSITIONS = {
+		"address": ParserState.CAPTURE,
+		"source": ParserState.CAPTURE,
+		"destination": ParserState.CAPTURE,
+		"global": ParserState.CAPTURE,
+	}
+
+	def __init__(self):
+		self._reset()
+
+	def _reset(self):
+		self._state = ParserState.NOOP
+		self._address = None
+		self._source = None
+		self._destination = None
+		self._global = None
+		self._content = ""
+
+	def _processElement(self, name):
+		pass
+
+	def startElement(self, name, attrs):
+		if name == "record":
+			self._state = ParserState.DISCARD
+			return
+		if self._state == ParserState.NOOP:
+			return
+
+		self._state = self.TRANSITIONS.get(name, ParserState.DISCARD)
+
+	def endElement(self, name):
+		if name == "record":
+			if self._address is not None:
+				self._processElement()
+			self._reset()
+		if self._state == ParserState.NOOP:
+			return
+
+		if self._content == "":
+			return
+		self._content = self._content.strip()
+		if name == "address":
+			self._address = self._content.split(", ")
+		if name == "source":
+			self._source = self._content == "True"
+		if name == "destination":
+			self._destination = self._content == "True"
+		if name == "global":
+			self._global = self._content == "True"
+		self._content = ""
+		self._state = ParserState.DISCARD
+
+	def characters(self, content):
+		if self._state != ParserState.CAPTURE:
+			return
+		self._content += content
+
+
+class IPDisallowedInboundHandler(IPDisallowedHandler):
+	def _processElement(self):
+		if not self._source:
+			for address in self._address:
+				print(address)
+
+
+class IPDisallowedOutboundHandler(IPDisallowedHandler):
+	def _processElement(self):
+		# Document states:
+		# > If the value of "Destination" is FALSE, the values of
+		# > "Forwardable" and "Globally Reachable" must also be false.
+		# So, `self._destination == False` implies `self._global ==
+		# False.` For this reason, it's enough to test for
+		# `self._global == False`.
+		if not self._global:
+			for address in self._address:
+				print(address)
 
 if __name__ == "__main__":
-	def usage():
-		print("Usage: {}".format(sys.argv[1]), file=sys.stderr)
-		sys.exit(1)
+	parser = argparse.ArgumentParser()
+	parser.add_argument("direction", choices=["inbound", "outbound"])
+	args = parser.parse_args()
 
-	if len(sys.argv) != 1:
-		usage()
-
-	parse_iana_csv(sys.stdin)
+	if args.direction == "inbound":
+		handler = IPDisallowedInboundHandler()
+	else:
+		handler = IPDisallowedOutboundHandler()
+	xml.sax.parse(sys.stdin, handler)