diff options
Diffstat (limited to 'rp/rpki-rtr')
| -rw-r--r-- | rp/rpki-rtr/Makefile.in | 51 | ||||
| -rw-r--r-- | rp/rpki-rtr/README | 11 | ||||
| -rwxr-xr-x | rp/rpki-rtr/rtr-origin | 2278 | ||||
| -rw-r--r-- | rp/rpki-rtr/rules.darwin.mk | 9 | ||||
| -rw-r--r-- | rp/rpki-rtr/rules.freebsd.mk | 37 | ||||
| -rw-r--r-- | rp/rpki-rtr/rules.linux.mk | 29 | ||||
| -rw-r--r-- | rp/rpki-rtr/rules.unknown.mk | 8 | ||||
| -rwxr-xr-x | rp/rpki-rtr/server.sh | 17 | ||||
| -rw-r--r-- | rp/rpki-rtr/sshd.conf | 23 |
9 files changed, 2463 insertions, 0 deletions
diff --git a/rp/rpki-rtr/Makefile.in b/rp/rpki-rtr/Makefile.in new file mode 100644 index 00000000..e587305b --- /dev/null +++ b/rp/rpki-rtr/Makefile.in @@ -0,0 +1,51 @@ +# $Id$ + +BIN = rtr-origin + +INSTALL = @INSTALL@ +PYTHON = @PYTHON@ +AWK = @AWK@ + +prefix = @prefix@ +exec_prefix = @exec_prefix@ +datarootdir = @datarootdir@ +datadir = @datadir@ +localstatedir = @localstatedir@ +sharedstatedir = @sharedstatedir@ +sysconfdir = @sysconfdir@ +bindir = @bindir@ +sbindir = @sbindir@ +libexecdir = @libexecdir@ +libdir = @libdir@ + +abs_top_srcdir = @abs_top_srcdir@ +abs_top_builddir = @abs_top_builddir@ + +RTR_ORIGIN_INSTALL_TARGETS = @RCYNIC_INSTALL_TARGETS@ + +RPKI_RTR_PORT = 43779 + +SCAN_ROAS = ${bindir}/scan_roas + + +all clean test:: + @true + +install: all ${RTR_ORIGIN_INSTALL_TARGETS} + +install-binary: + if test -d ${DESTDIR}${bindir} ; then :; else ${INSTALL} -m 555 -d ${DESTDIR}${bindir}; fi + ${INSTALL} -m 555 ${BIN} ${DESTDIR}${bindir}/${BIN} + +deinstall uninstall: + rm -f ${DESTDIR}${bindir}/${BIN} + +distclean: clean + rm -rf current sockets *.ax *.ix.* + rm -f Makefile + +.FORCE: + +# Platform-specific rules below here. + +@RTR_ORIGIN_MAKE_RULES@ diff --git a/rp/rpki-rtr/README b/rp/rpki-rtr/README new file mode 100644 index 00000000..dae53010 --- /dev/null +++ b/rp/rpki-rtr/README @@ -0,0 +1,11 @@ +$Id$ + +Sample implementation of draft-ymbk-rpki-rtr-protocol. + +See: + +- The primary documentation at http://trac.rpki.net/ + +- The PDF manual in ../doc/manual.pdf, or + +- The flat text page ../doc/doc.RPKI.RP.rpki-rtr diff --git a/rp/rpki-rtr/rtr-origin b/rp/rpki-rtr/rtr-origin new file mode 100755 index 00000000..e1e82ccf --- /dev/null +++ b/rp/rpki-rtr/rtr-origin @@ -0,0 +1,2278 @@ +#!/usr/bin/env python + +# Router origin-authentication rpki-router protocol implementation. See +# draft-ietf-sidr-rpki-rtr in fine Internet-Draft repositories near you. +# +# Run the program with the --help argument for usage information, or see +# documentation for the *_main() functions. +# +# +# $Id$ +# +# Copyright (C) 2009-2013 Internet Systems Consortium ("ISC") +# +# Permission to use, copy, modify, and distribute this software for any +# purpose with or without fee is hereby granted, provided that the above +# copyright notice and this permission notice appear in all copies. +# +# THE SOFTWARE IS PROVIDED "AS IS" AND ISC DISCLAIMS ALL WARRANTIES WITH +# REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY +# AND FITNESS. IN NO EVENT SHALL ISC BE LIABLE FOR ANY SPECIAL, DIRECT, +# INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM +# LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE +# OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR +# PERFORMANCE OF THIS SOFTWARE. + +import sys +import os +import struct +import time +import glob +import socket +import fcntl +import signal +import syslog +import errno +import asyncore +import asynchat +import subprocess +import traceback +import getopt +import bisect +import random +import base64 + + +# Debugging only, should be False in production +disable_incrementals = False + +# Whether to log backtraces +backtrace_on_exceptions = False + +class IgnoreThisRecord(Exception): + pass + + +class timestamp(int): + """ + Wrapper around time module. + """ + + def __new__(cls, x): + return int.__new__(cls, x) + + @classmethod + def now(cls, delta = 0): + return cls(time.time() + delta) + + def __str__(self): + return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(self)) + + +class ipaddr(object): + """ + IP addresses. + """ + + def __init__(self, string = None, value = None): + assert (string is None) != (value is None) + if string is not None: + value = socket.inet_pton(self.af, string) + assert len(value) == self.size + self.value = value + + def __str__(self): + return socket.inet_ntop(self.af, self.value) + + def __cmp__(self, other): + return cmp(self.value, other.value) + +class v4addr(ipaddr): + af = socket.AF_INET + size = 4 + +class v6addr(ipaddr): + af = socket.AF_INET6 + size = 16 + +def read_current(): + """ + Read current serial number and nonce. Return None for both if + serial and nonce not recorded. For backwards compatibility, treat + file containing just a serial number as having a nonce of zero. + """ + try: + f = open("current", "r") + values = tuple(int(s) for s in f.read().split()) + f.close() + return values[0], values[1] + except IndexError: + return values[0], 0 + except IOError: + return None, None + +def write_current(serial, nonce): + """ + Write serial number and nonce. + """ + tmpfn = "current.%d.tmp" % os.getpid() + try: + f = open(tmpfn, "w") + f.write("%d %d\n" % (serial, nonce)) + f.close() + os.rename(tmpfn, "current") + finally: + try: + os.unlink(tmpfn) + except: + pass + + +def new_nonce(): + """ + Create and return a new nonce value. + """ + if force_zero_nonce: + return 0 + try: + return int(random.SystemRandom().getrandbits(16)) + except NotImplementedError: + return int(random.getrandbits(16)) + + +class read_buffer(object): + """ + Wrapper around synchronous/asynchronous read state. + """ + + def __init__(self): + self.buffer = "" + + def update(self, need, callback): + """ + Update count of needed bytes and callback, then dispatch to callback. + """ + self.need = need + self.callback = callback + return self.callback(self) + + def available(self): + """ + How much data do we have available in this buffer? + """ + return len(self.buffer) + + def needed(self): + """ + How much more data does this buffer need to become ready? + """ + return self.need - self.available() + + def ready(self): + """ + Is this buffer ready to read yet? + """ + return self.available() >= self.need + + def get(self, n): + """ + Hand some data to the caller. + """ + b = self.buffer[:n] + self.buffer = self.buffer[n:] + return b + + def put(self, b): + """ + Accumulate some data. + """ + self.buffer += b + + def retry(self): + """ + Try dispatching to the callback again. + """ + return self.callback(self) + +class PDUException(Exception): + """ + Parent exception type for exceptions that signal particular protocol + errors. String value of exception instance will be the message to + put in the error_report PDU, error_report_code value of exception + will be the numeric code to use. + """ + + def __init__(self, msg = None, pdu = None): + assert msg is None or isinstance(msg, (str, unicode)) + self.error_report_msg = msg + self.error_report_pdu = pdu + + def __str__(self): + return self.error_report_msg or self.__class__.__name__ + + def make_error_report(self): + return error_report(errno = self.error_report_code, + errmsg = self.error_report_msg, + errpdu = self.error_report_pdu) + +class UnsupportedProtocolVersion(PDUException): + error_report_code = 4 + +class UnsupportedPDUType(PDUException): + error_report_code = 5 + +class CorruptData(PDUException): + error_report_code = 0 + +class pdu(object): + """ + Object representing a generic PDU in the rpki-router protocol. + Real PDUs are subclasses of this class. + """ + + version = 0 # Protocol version + + _pdu = None # Cached when first generated + + header_struct = struct.Struct("!BBHL") + + def __cmp__(self, other): + return cmp(self.to_pdu(), other.to_pdu()) + + def check(self): + """ + Check attributes to make sure they're within range. + """ + pass + + @classmethod + def read_pdu(cls, reader): + return reader.update(need = cls.header_struct.size, callback = cls.got_header) + + @classmethod + def got_header(cls, reader): + if not reader.ready(): + return None + assert reader.available() >= cls.header_struct.size + version, pdu_type, whatever, length = cls.header_struct.unpack(reader.buffer[:cls.header_struct.size]) + if version != cls.version: + raise UnsupportedProtocolVersion( + "Received PDU version %d, expected %d" % (version, cls.version)) + if pdu_type not in cls.pdu_map: + raise UnsupportedPDUType( + "Received unsupported PDU type %d" % pdu_type) + if length < 8: + raise CorruptData( + "Received PDU with length %d, which is too short to be valid" % length) + self = cls.pdu_map[pdu_type]() + return reader.update(need = length, callback = self.got_pdu) + + def consume(self, client): + """ + Handle results in test client. Default behavior is just to print + out the PDU. + """ + blather(self) + + def send_file(self, server, filename): + """ + Send a content of a file as a cache response. Caller should catch IOError. + """ + f = open(filename, "rb") + server.push_pdu(cache_response(nonce = server.current_nonce)) + server.push_file(f) + server.push_pdu(end_of_data(serial = server.current_serial, nonce = server.current_nonce)) + + def send_nodata(self, server): + """ + Send a nodata error. + """ + server.push_pdu(error_report(errno = error_report.codes["No Data Available"], errpdu = self)) + +class pdu_with_serial(pdu): + """ + Base class for PDUs consisting of just a serial number and nonce. + """ + + header_struct = struct.Struct("!BBHLL") + + def __init__(self, serial = None, nonce = None): + if serial is not None: + assert isinstance(serial, int) + self.serial = serial + if nonce is not None: + assert isinstance(nonce, int) + self.nonce = nonce + + def __str__(self): + return "[%s, serial #%d nonce %d]" % (self.__class__.__name__, self.serial, self.nonce) + + def to_pdu(self): + """ + Generate the wire format PDU. + """ + if self._pdu is None: + self._pdu = self.header_struct.pack(self.version, self.pdu_type, self.nonce, + self.header_struct.size, self.serial) + return self._pdu + + def got_pdu(self, reader): + if not reader.ready(): + return None + b = reader.get(self.header_struct.size) + version, pdu_type, self.nonce, length, self.serial = self.header_struct.unpack(b) + if length != 12: + raise CorruptData("PDU length of %d can't be right" % length, pdu = self) + assert b == self.to_pdu() + return self + +class pdu_nonce(pdu): + """ + Base class for PDUs consisting of just a nonce. + """ + + header_struct = struct.Struct("!BBHL") + + def __init__(self, nonce = None): + if nonce is not None: + assert isinstance(nonce, int) + self.nonce = nonce + + def __str__(self): + return "[%s, nonce %d]" % (self.__class__.__name__, self.nonce) + + def to_pdu(self): + """ + Generate the wire format PDU. + """ + if self._pdu is None: + self._pdu = self.header_struct.pack(self.version, self.pdu_type, self.nonce, self.header_struct.size) + return self._pdu + + def got_pdu(self, reader): + if not reader.ready(): + return None + b = reader.get(self.header_struct.size) + version, pdu_type, self.nonce, length = self.header_struct.unpack(b) + if length != 8: + raise CorruptData("PDU length of %d can't be right" % length, pdu = self) + assert b == self.to_pdu() + return self + +class pdu_empty(pdu): + """ + Base class for empty PDUs. + """ + + header_struct = struct.Struct("!BBHL") + + def __str__(self): + return "[%s]" % self.__class__.__name__ + + def to_pdu(self): + """ + Generate the wire format PDU for this prefix. + """ + if self._pdu is None: + self._pdu = self.header_struct.pack(self.version, self.pdu_type, 0, self.header_struct.size) + return self._pdu + + def got_pdu(self, reader): + if not reader.ready(): + return None + b = reader.get(self.header_struct.size) + version, pdu_type, zero, length = self.header_struct.unpack(b) + if zero != 0: + raise CorruptData("Must-be-zero field isn't zero" % length, pdu = self) + if length != 8: + raise CorruptData("PDU length of %d can't be right" % length, pdu = self) + assert b == self.to_pdu() + return self + +class serial_notify(pdu_with_serial): + """ + Serial Notify PDU. + """ + + pdu_type = 0 + + def consume(self, client): + """ + Respond to a serial_notify message with either a serial_query or + reset_query, depending on what we already know. + """ + blather(self) + if client.current_serial is None or client.current_nonce != self.nonce: + client.push_pdu(reset_query()) + elif self.serial != client.current_serial: + client.push_pdu(serial_query(serial = client.current_serial, nonce = client.current_nonce)) + else: + blather("[Notify did not change serial number, ignoring]") + +class serial_query(pdu_with_serial): + """ + Serial Query PDU. + """ + + pdu_type = 1 + + def serve(self, server): + """ + Received a serial query, send incremental transfer in response. + If client is already up to date, just send an empty incremental + transfer. + """ + blather(self) + if server.get_serial() is None: + self.send_nodata(server) + elif server.current_nonce != self.nonce: + log("[Client requested wrong nonce, resetting client]") + server.push_pdu(cache_reset()) + elif server.current_serial == self.serial: + blather("[Client is already current, sending empty IXFR]") + server.push_pdu(cache_response(nonce = server.current_nonce)) + server.push_pdu(end_of_data(serial = server.current_serial, nonce = server.current_nonce)) + elif disable_incrementals: + server.push_pdu(cache_reset()) + else: + try: + self.send_file(server, "%d.ix.%d" % (server.current_serial, self.serial)) + except IOError: + server.push_pdu(cache_reset()) + +class reset_query(pdu_empty): + """ + Reset Query PDU. + """ + + pdu_type = 2 + + def serve(self, server): + """ + Received a reset query, send full current state in response. + """ + blather(self) + if server.get_serial() is None: + self.send_nodata(server) + else: + try: + fn = "%d.ax" % server.current_serial + self.send_file(server, fn) + except IOError: + server.push_pdu(error_report(errno = error_report.codes["Internal Error"], + errpdu = self, errmsg = "Couldn't open %s" % fn)) + +class cache_response(pdu_nonce): + """ + Cache Response PDU. + """ + + pdu_type = 3 + + def consume(self, client): + """ + Handle cache_response. + """ + blather(self) + if self.nonce != client.current_nonce: + blather("[Nonce changed, resetting]") + client.cache_reset() + +class end_of_data(pdu_with_serial): + """ + End of Data PDU. + """ + + pdu_type = 7 + + def consume(self, client): + """ + Handle end_of_data response. + """ + blather(self) + client.end_of_data(self.serial, self.nonce) + +class cache_reset(pdu_empty): + """ + Cache reset PDU. + """ + + pdu_type = 8 + + def consume(self, client): + """ + Handle cache_reset response, by issuing a reset_query. + """ + blather(self) + client.cache_reset() + client.push_pdu(reset_query()) + +class prefix(pdu): + """ + Object representing one prefix. This corresponds closely to one PDU + in the rpki-router protocol, so closely that we use lexical ordering + of the wire format of the PDU as the ordering for this class. + + This is a virtual class, but the .from_text() constructor + instantiates the correct concrete subclass (ipv4_prefix or + ipv6_prefix) depending on the syntax of its input text. + """ + + header_struct = struct.Struct("!BB2xLBBBx") + asnum_struct = struct.Struct("!L") + + @staticmethod + def from_text(asnum, addr): + """ + Construct a prefix from its text form. + """ + cls = ipv6_prefix if ":" in addr else ipv4_prefix + self = cls() + self.asn = long(asnum) + p, l = addr.split("/") + self.prefix = self.addr_type(string = p) + if "-" in l: + self.prefixlen, self.max_prefixlen = tuple(int(i) for i in l.split("-")) + else: + self.prefixlen = self.max_prefixlen = int(l) + self.announce = 1 + self.check() + return self + + def __str__(self): + plm = "%s/%s-%s" % (self.prefix, self.prefixlen, self.max_prefixlen) + return "%s %8s %-32s %s" % ("+" if self.announce else "-", self.asn, plm, + ":".join(("%02X" % ord(b) for b in self.to_pdu()))) + + def show(self): + blather("# Class: %s" % self.__class__.__name__) + blather("# ASN: %s" % self.asn) + blather("# Prefix: %s" % self.prefix) + blather("# Prefixlen: %s" % self.prefixlen) + blather("# MaxPrefixlen: %s" % self.max_prefixlen) + blather("# Announce: %s" % self.announce) + + def consume(self, client): + """ + Handle one incoming prefix PDU + """ + blather(self) + client.consume_prefix(self) + + def check(self): + """ + Check attributes to make sure they're within range. + """ + if self.announce not in (0, 1): + raise CorruptData("Announce value %d is neither zero nor one" % self.announce, pdu = self) + if self.prefixlen < 0 or self.prefixlen > self.addr_type.size * 8: + raise CorruptData("Implausible prefix length %d" % self.prefixlen, pdu = self) + if self.max_prefixlen < self.prefixlen or self.max_prefixlen > self.addr_type.size * 8: + raise CorruptData("Implausible max prefix length %d" % self.max_prefixlen, pdu = self) + pdulen = self.header_struct.size + self.addr_type.size + self.asnum_struct.size + if len(self.to_pdu()) != pdulen: + raise CorruptData("Expected %d byte PDU, got %d" % (pdulen, len(self.to_pdu())), pdu = self) + + def to_pdu(self, announce = None): + """ + Generate the wire format PDU for this prefix. + """ + if announce is not None: + assert announce in (0, 1) + elif self._pdu is not None: + return self._pdu + pdulen = self.header_struct.size + self.addr_type.size + self.asnum_struct.size + pdu = (self.header_struct.pack(self.version, self.pdu_type, pdulen, + announce if announce is not None else self.announce, + self.prefixlen, self.max_prefixlen) + + self.prefix.value + + self.asnum_struct.pack(self.asn)) + if announce is None: + assert self._pdu is None + self._pdu = pdu + return pdu + + def got_pdu(self, reader): + if not reader.ready(): + return None + b1 = reader.get(self.header_struct.size) + b2 = reader.get(self.addr_type.size) + b3 = reader.get(self.asnum_struct.size) + version, pdu_type, length, self.announce, self.prefixlen, self.max_prefixlen = self.header_struct.unpack(b1) + if length != len(b1) + len(b2) + len(b3): + raise CorruptData("Got PDU length %d, expected %d" % (length, len(b1) + len(b2) + len(b3)), pdu = self) + self.prefix = self.addr_type(value = b2) + self.asn = self.asnum_struct.unpack(b3)[0] + assert b1 + b2 + b3 == self.to_pdu() + return self + + @staticmethod + def from_bgpdump(line, rib_dump): + try: + assert isinstance(rib_dump, bool) + fields = line.split("|") + + # Parse prefix, including figuring out IP protocol version + cls = ipv6_prefix if ":" in fields[5] else ipv4_prefix + self = cls() + self.timestamp = timestamp(fields[1]) + p, l = fields[5].split("/") + self.prefix = self.addr_type(p) + self.prefixlen = self.max_prefixlen = int(l) + + # Withdrawals don't have AS paths, so be careful + assert fields[2] == "B" if rib_dump else fields[2] in ("A", "W") + if fields[2] == "W": + self.asn = 0 + self.announce = 0 + else: + self.announce = 1 + if not fields[6] or "{" in fields[6] or "(" in fields[6]: + raise IgnoreThisRecord + a = fields[6].split()[-1] + if "." in a: + a = [int(s) for s in a.split(".")] + if len(a) != 2 or a[0] < 0 or a[0] > 65535 or a[1] < 0 or a[1] > 65535: + log("Bad dotted ASNum %r, ignoring record" % fields[6]) + raise IgnoreThisRecord + a = (a[0] << 16) | a[1] + else: + a = int(a) + self.asn = a + + self.check() + return self + + except IgnoreThisRecord: + raise + + except Exception, e: + log("Ignoring line %r: %s" % (line, e)) + raise IgnoreThisRecord + +class ipv4_prefix(prefix): + """ + IPv4 flavor of a prefix. + """ + pdu_type = 4 + addr_type = v4addr + +class ipv6_prefix(prefix): + """ + IPv6 flavor of a prefix. + """ + pdu_type = 6 + addr_type = v6addr + +class router_key(pdu): + """ + Router Key PDU. + """ + + pdu_type = 9 + + header_struct = struct.Struct("!BBBxL20sL") + + @classmethod + def from_text(cls, asnum, gski, key): + """ + Construct a router key from its text form. + """ + + self = cls() + self.asn = long(asnum) + self.ski = base64.urlsafe_b64decode(gski + "=") + self.key = base64.b64decode(key) + self.announce = 1 + self.check() + return self + + def __str__(self): + return "%s %8s %-32s %s" % ("+" if self.announce else "-", self.asn, + base64.urlsafe_b64encode(self.ski).rstrip("="), + ":".join(("%02X" % ord(b) for b in self.to_pdu()))) + + def consume(self, client): + """ + Handle one incoming Router Key PDU + """ + + blather(self) + client.consume_routerkey(self) + + def check(self): + """ + Check attributes to make sure they're within range. + """ + + if self.announce not in (0, 1): + raise CorruptData("Announce value %d is neither zero nor one" % self.announce, pdu = self) + if len(self.ski) != 20: + raise CorruptData("Implausible SKI length %d" % len(self.ski), pdu = self) + pdulen = self.header_struct.size + len(self.key) + if len(self.to_pdu()) != pdulen: + raise CorruptData("Expected %d byte PDU, got %d" % (pdulen, len(self.to_pdu())), pdu = self) + + def to_pdu(self, announce = None): + if announce is not None: + assert announce in (0, 1) + elif self._pdu is not None: + return self._pdu + pdulen = self.header_struct.size + len(self.key) + pdu = (self.header_struct.pack(self.version, + self.pdu_type, + announce if announce is not None else self.announce, + pdulen, + self.ski, + self.asn) + + self.key) + if announce is None: + assert self._pdu is None + self._pdu = pdu + return pdu + + def got_pdu(self, reader): + if not reader.ready(): + return None + header = reader.get(self.header_struct.size) + version, pdu_type, self.announce, length, self.ski, self.asn = self.header_struct.unpack(header) + remaining = length - self.header_struct.size + if remaining <= 0: + raise CorruptData("Got PDU length %d, minimum is %d" % (length, self.header_struct.size + 1), pdu = self) + self.key = reader.get(remaining) + assert header + self.key == self.to_pdu() + return self + + +class error_report(pdu): + """ + Error Report PDU. + """ + + pdu_type = 10 + + header_struct = struct.Struct("!BBHL") + string_struct = struct.Struct("!L") + + errors = { + 2 : "No Data Available" } + + fatal = { + 0 : "Corrupt Data", + 1 : "Internal Error", + 3 : "Invalid Request", + 4 : "Unsupported Protocol Version", + 5 : "Unsupported PDU Type", + 6 : "Withdrawal of Unknown Record", + 7 : "Duplicate Announcement Received" } + + assert set(errors) & set(fatal) == set() + + errors.update(fatal) + + codes = dict((v, k) for k, v in errors.items()) + + def __init__(self, errno = None, errpdu = None, errmsg = None): + assert errno is None or errno in self.errors + self.errno = errno + self.errpdu = errpdu + self.errmsg = errmsg if errmsg is not None or errno is None else self.errors[errno] + + def __str__(self): + return "[%s, error #%s: %r]" % (self.__class__.__name__, self.errno, self.errmsg) + + def to_counted_string(self, s): + return self.string_struct.pack(len(s)) + s + + def read_counted_string(self, reader, remaining): + assert remaining >= self.string_struct.size + n = self.string_struct.unpack(reader.get(self.string_struct.size))[0] + assert remaining >= self.string_struct.size + n + return n, reader.get(n), (remaining - self.string_struct.size - n) + + def to_pdu(self): + """ + Generate the wire format PDU for this error report. + """ + if self._pdu is None: + assert isinstance(self.errno, int) + assert not isinstance(self.errpdu, error_report) + p = self.errpdu + if p is None: + p = "" + elif isinstance(p, pdu): + p = p.to_pdu() + assert isinstance(p, str) + pdulen = self.header_struct.size + self.string_struct.size * 2 + len(p) + len(self.errmsg) + self._pdu = self.header_struct.pack(self.version, self.pdu_type, self.errno, pdulen) + self._pdu += self.to_counted_string(p) + self._pdu += self.to_counted_string(self.errmsg.encode("utf8")) + return self._pdu + + def got_pdu(self, reader): + if not reader.ready(): + return None + header = reader.get(self.header_struct.size) + version, pdu_type, self.errno, length = self.header_struct.unpack(header) + remaining = length - self.header_struct.size + self.pdulen, self.errpdu, remaining = self.read_counted_string(reader, remaining) + self.errlen, self.errmsg, remaining = self.read_counted_string(reader, remaining) + if length != self.header_struct.size + self.string_struct.size * 2 + self.pdulen + self.errlen: + raise CorruptData("Got PDU length %d, expected %d" % ( + length, self.header_struct.size + self.string_struct.size * 2 + self.pdulen + self.errlen)) + assert (header + + self.to_counted_string(self.errpdu) + + self.to_counted_string(self.errmsg.encode("utf8")) + == self.to_pdu()) + return self + + def serve(self, server): + """ + Received an error_report from client. Not much we can do beyond + logging it, then killing the connection if error was fatal. + """ + log(self) + if self.errno in self.fatal: + log("[Shutting down due to reported fatal protocol error]") + sys.exit(1) + +pdu.pdu_map = dict((p.pdu_type, p) for p in (ipv4_prefix, ipv6_prefix, serial_notify, serial_query, reset_query, + cache_response, end_of_data, cache_reset, router_key, error_report)) + +class pdu_set(list): + """ + Object representing a set of PDUs, that is, one versioned and + (theoretically) consistant set of prefixes and router keys extracted + from rcynic's output. + """ + + @classmethod + def _load_file(cls, filename): + """ + Low-level method to read pdu_set from a file. + """ + self = cls() + f = open(filename, "rb") + r = read_buffer() + while True: + p = pdu.read_pdu(r) + while p is None: + b = f.read(r.needed()) + if b == "": + assert r.available() == 0 + return self + r.put(b) + p = r.retry() + self.append(p) + + @staticmethod + def seq_ge(a, b): + return ((a - b) % (1 << 32)) < (1 << 31) + + +class axfr_set(pdu_set): + """ + Object representing a complete set of PDUs, that is, one versioned + and (theoretically) consistant set of prefixes and router + certificates extracted from rcynic's output, all with the announce + field set. + """ + + @classmethod + def parse_rcynic(cls, rcynic_dir): + """ + Parse ROAS and router certificates fetched (and validated!) by + rcynic to create a new axfr_set. We use the scan_roas and + scan_routercerts utilities to parse the ASN.1, although we may go + back to parsing the files directly using the rpki.POW library code + some day. + """ + + self = cls() + self.serial = timestamp.now() + + try: + p = subprocess.Popen((scan_roas, rcynic_dir), stdout = subprocess.PIPE) + for line in p.stdout: + line = line.split() + asn = line[1] + self.extend(prefix.from_text(asn, addr) for addr in line[2:]) + except OSError, e: + sys.exit("Could not run %s, check your $PATH variable? (%s)" % (scan_roas, e)) + + try: + p = subprocess.Popen((scan_routercerts, rcynic_dir), stdout = subprocess.PIPE) + for line in p.stdout: + line = line.split() + gski = line[0] + key = line[-1] + self.extend(router_key.from_text(asn, gski, key) for asn in line[1:-1]) + except OSError, e: + sys.exit("Could not run %s, check your $PATH variable? (%s)" % (scan_routercerts, e)) + + self.sort() + for i in xrange(len(self) - 2, -1, -1): + if self[i] == self[i + 1]: + del self[i + 1] + return self + + @classmethod + def load(cls, filename): + """ + Load an axfr_set from a file, parse filename to obtain serial. + """ + fn1, fn2 = os.path.basename(filename).split(".") + assert fn1.isdigit() and fn2 == "ax" + self = cls._load_file(filename) + self.serial = timestamp(fn1) + return self + + def filename(self): + """ + Generate filename for this axfr_set. + """ + return "%d.ax" % self.serial + + @classmethod + def load_current(cls): + """ + Load current axfr_set. Return None if can't. + """ + serial = read_current()[0] + if serial is None: + return None + try: + return cls.load("%d.ax" % serial) + except IOError: + return None + + def save_axfr(self): + """ + Write axfr__set to file with magic filename. + """ + f = open(self.filename(), "wb") + for p in self: + f.write(p.to_pdu()) + f.close() + + def destroy_old_data(self): + """ + Destroy old data files, presumably because our nonce changed and + the old serial numbers are no longer valid. + """ + for i in glob.iglob("*.ix.*"): + os.unlink(i) + for i in glob.iglob("*.ax"): + if i != self.filename(): + os.unlink(i) + + def mark_current(self): + """ + Save current serial number and nonce, creating new nonce if + necessary. Creating a new nonce triggers cleanup of old state, as + the new nonce invalidates all old serial numbers. + """ + old_serial, nonce = read_current() + if old_serial is None or self.seq_ge(old_serial, self.serial): + blather("Creating new nonce and deleting stale data") + nonce = new_nonce() + self.destroy_old_data() + write_current(self.serial, nonce) + + def save_ixfr(self, other): + """ + Comparing this axfr_set with an older one and write the resulting + ixfr_set to file with magic filename. Since we store pdu_sets + in sorted order, computing the difference is a trivial linear + comparison. + """ + f = open("%d.ix.%d" % (self.serial, other.serial), "wb") + old = other + new = self + len_old = len(old) + len_new = len(new) + i_old = i_new = 0 + while i_old < len_old and i_new < len_new: + if old[i_old] < new[i_new]: + f.write(old[i_old].to_pdu(announce = 0)) + i_old += 1 + elif old[i_old] > new[i_new]: + f.write(new[i_new].to_pdu(announce = 1)) + i_new += 1 + else: + i_old += 1 + i_new += 1 + for i in xrange(i_old, len_old): + f.write(old[i].to_pdu(announce = 0)) + for i in xrange(i_new, len_new): + f.write(new[i].to_pdu(announce = 1)) + f.close() + + def show(self): + """ + Print this axfr_set. + """ + blather("# AXFR %d (%s)" % (self.serial, self.serial)) + for p in self: + blather(p) + + @staticmethod + def read_bgpdump(filename): + assert filename.endswith(".bz2") + blather("Reading %s" % filename) + bunzip2 = subprocess.Popen(("bzip2", "-c", "-d", filename), stdout = subprocess.PIPE) + bgpdump = subprocess.Popen(("bgpdump", "-m", "-"), stdin = bunzip2.stdout, stdout = subprocess.PIPE) + return bgpdump.stdout + + @classmethod + def parse_bgpdump_rib_dump(cls, filename): + assert os.path.basename(filename).startswith("ribs.") + self = cls() + self.serial = None + for line in cls.read_bgpdump(filename): + try: + pfx = prefix.from_bgpdump(line, rib_dump = True) + except IgnoreThisRecord: + continue + self.append(pfx) + self.serial = pfx.timestamp + if self.serial is None: + sys.exit("Failed to parse anything useful from %s" % filename) + self.sort() + for i in xrange(len(self) - 2, -1, -1): + if self[i] == self[i + 1]: + del self[i + 1] + return self + + def parse_bgpdump_update(self, filename): + assert os.path.basename(filename).startswith("updates.") + for line in self.read_bgpdump(filename): + try: + pfx = prefix.from_bgpdump(line, rib_dump = False) + except IgnoreThisRecord: + continue + announce = pfx.announce + pfx.announce = 1 + i = bisect.bisect_left(self, pfx) + if announce: + if i >= len(self) or pfx != self[i]: + self.insert(i, pfx) + else: + while i < len(self) and pfx.prefix == self[i].prefix and pfx.prefixlen == self[i].prefixlen: + del self[i] + self.serial = pfx.timestamp + +class ixfr_set(pdu_set): + """ + Object representing an incremental set of PDUs, that is, the + differences between one versioned and (theoretically) consistant set + of prefixes and router certificates extracted from rcynic's output + and another, with the announce fields set or cleared as necessary to + indicate the changes. + """ + + @classmethod + def load(cls, filename): + """ + Load an ixfr_set from a file, parse filename to obtain serials. + """ + fn1, fn2, fn3 = os.path.basename(filename).split(".") + assert fn1.isdigit() and fn2 == "ix" and fn3.isdigit() + self = cls._load_file(filename) + self.from_serial = timestamp(fn3) + self.to_serial = timestamp(fn1) + return self + + def filename(self): + """ + Generate filename for this ixfr_set. + """ + return "%d.ix.%d" % (self.to_serial, self.from_serial) + + def show(self): + """ + Print this ixfr_set. + """ + blather("# IXFR %d (%s) -> %d (%s)" % (self.from_serial, self.from_serial, + self.to_serial, self.to_serial)) + for p in self: + blather(p) + +class file_producer(object): + """ + File-based producer object for asynchat. + """ + + def __init__(self, handle, buffersize): + self.handle = handle + self.buffersize = buffersize + + def more(self): + return self.handle.read(self.buffersize) + +class pdu_channel(asynchat.async_chat): + """ + asynchat subclass that understands our PDUs. This just handles + network I/O. Specific engines (client, server) should be subclasses + of this with methods that do something useful with the resulting + PDUs. + """ + + def __init__(self, conn = None): + asynchat.async_chat.__init__(self, conn) + self.reader = read_buffer() + + def start_new_pdu(self): + """ + Start read of a new PDU. + """ + try: + p = pdu.read_pdu(self.reader) + while p is not None: + self.deliver_pdu(p) + p = pdu.read_pdu(self.reader) + except PDUException, e: + self.push_pdu(e.make_error_report()) + self.close_when_done() + else: + assert not self.reader.ready() + self.set_terminator(self.reader.needed()) + + def collect_incoming_data(self, data): + """ + Collect data into the read buffer. + """ + self.reader.put(data) + + def found_terminator(self): + """ + Got requested data, see if we now have a PDU. If so, pass it + along, then restart cycle for a new PDU. + """ + p = self.reader.retry() + if p is None: + self.set_terminator(self.reader.needed()) + else: + self.deliver_pdu(p) + self.start_new_pdu() + + def push_pdu(self, pdu): + """ + Write PDU to stream. + """ + try: + self.push(pdu.to_pdu()) + except OSError, e: + if e.errno != errno.EAGAIN: + raise + + def push_file(self, f): + """ + Write content of a file to stream. + """ + try: + self.push_with_producer(file_producer(f, self.ac_out_buffer_size)) + except OSError, e: + if e.errno != errno.EAGAIN: + raise + + def log(self, msg): + """ + Intercept asyncore's logging. + """ + log(msg) + + def log_info(self, msg, tag = "info"): + """ + Intercept asynchat's logging. + """ + log("asynchat: %s: %s" % (tag, msg)) + + def handle_error(self): + """ + Handle errors caught by asyncore main loop. + """ + c, e = sys.exc_info()[:2] + if backtrace_on_exceptions or e == 0: + for line in traceback.format_exc().splitlines(): + log(line) + else: + log("[Exception: %s: %s]" % (c.__name__, e)) + log("[Exiting after unhandled exception]") + sys.exit(1) + + def init_file_dispatcher(self, fd): + """ + Kludge to plug asyncore.file_dispatcher into asynchat. Call from + subclass's __init__() method, after calling + pdu_channel.__init__(), and don't read this on a full stomach. + """ + self.connected = True + self._fileno = fd + self.socket = asyncore.file_wrapper(fd) + self.add_channel() + flags = fcntl.fcntl(fd, fcntl.F_GETFL, 0) + flags = flags | os.O_NONBLOCK + fcntl.fcntl(fd, fcntl.F_SETFL, flags) + + def handle_close(self): + """ + Exit when channel closed. + """ + asynchat.async_chat.handle_close(self) + sys.exit(0) + +class server_write_channel(pdu_channel): + """ + Kludge to deal with ssh's habit of sometimes (compile time option) + invoking us with two unidirectional pipes instead of one + bidirectional socketpair. All the server logic is in the + server_channel class, this class just deals with sending the + server's output to a different file descriptor. + """ + + def __init__(self): + """ + Set up stdout. + """ + pdu_channel.__init__(self) + self.init_file_dispatcher(sys.stdout.fileno()) + + def readable(self): + """ + This channel is never readable. + """ + return False + +class server_channel(pdu_channel): + """ + Server protocol engine, handles upcalls from pdu_channel to + implement protocol logic. + """ + + def __init__(self): + """ + Set up stdin and stdout as connection and start listening for + first PDU. + """ + pdu_channel.__init__(self) + self.init_file_dispatcher(sys.stdin.fileno()) + self.writer = server_write_channel() + self.get_serial() + self.start_new_pdu() + + def writable(self): + """ + This channel is never writable. + """ + return False + + def push(self, data): + """ + Redirect to writer channel. + """ + return self.writer.push(data) + + def push_with_producer(self, producer): + """ + Redirect to writer channel. + """ + return self.writer.push_with_producer(producer) + + def push_pdu(self, pdu): + """ + Redirect to writer channel. + """ + return self.writer.push_pdu(pdu) + + def push_file(self, f): + """ + Redirect to writer channel. + """ + return self.writer.push_file(f) + + def deliver_pdu(self, pdu): + """ + Handle received PDU. + """ + pdu.serve(self) + + def get_serial(self): + """ + Read, cache, and return current serial number, or None if we can't + find the serial number file. The latter condition should never + happen, but maybe we got started in server mode while the cronjob + mode instance is still building its database. + """ + self.current_serial, self.current_nonce = read_current() + return self.current_serial + + def check_serial(self): + """ + Check for a new serial number. + """ + old_serial = self.current_serial + return old_serial != self.get_serial() + + def notify(self, data = None): + """ + Cronjob instance kicked us, send a notify message. + """ + if self.check_serial() is not None: + self.push_pdu(serial_notify(serial = self.current_serial, nonce = self.current_nonce)) + else: + log("Cronjob kicked me without a valid current serial number") + +class client_channel(pdu_channel): + """ + Client protocol engine, handles upcalls from pdu_channel. + """ + + current_serial = None + current_nonce = None + sql = None + host = None + port = None + cache_id = None + + def __init__(self, sock, proc, killsig, host, port): + self.killsig = killsig + self.proc = proc + self.host = host + self.port = port + pdu_channel.__init__(self, conn = sock) + self.start_new_pdu() + + @classmethod + def ssh(cls, host, port): + """ + Set up ssh connection and start listening for first PDU. + """ + args = ("ssh", "-p", port, "-s", host, "rpki-rtr") + blather("[Running ssh: %s]" % " ".join(args)) + s = socket.socketpair() + return cls(sock = s[1], + proc = subprocess.Popen(args, executable = "/usr/bin/ssh", + stdin = s[0], stdout = s[0], close_fds = True), + killsig = signal.SIGKILL, + host = host, port = port) + + @classmethod + def tcp(cls, host, port): + """ + Set up TCP connection and start listening for first PDU. + """ + blather("[Starting raw TCP connection to %s:%s]" % (host, port)) + try: + addrinfo = socket.getaddrinfo(host, port, socket.AF_UNSPEC, socket.SOCK_STREAM) + except socket.error, e: + blather("[socket.getaddrinfo() failed: %s]" % e) + else: + for ai in addrinfo: + af, socktype, proto, cn, sa = ai + blather("[Trying addr %s port %s]" % sa[:2]) + try: + s = socket.socket(af, socktype, proto) + except socket.error, e: + blather("[socket.socket() failed: %s]" % e) + continue + try: + s.connect(sa) + except socket.error, e: + blather("[socket.connect() failed: %s]" % e) + s.close() + continue + return cls(sock = s, proc = None, killsig = None, + host = host, port = port) + sys.exit(1) + + @classmethod + def loopback(cls, host, port): + """ + Set up loopback connection and start listening for first PDU. + """ + s = socket.socketpair() + blather("[Using direct subprocess kludge for testing]") + argv = [sys.executable, sys.argv[0], "--server"] + if "--syslog" in sys.argv: + argv.extend(("--syslog", sys.argv[sys.argv.index("--syslog") + 1])) + return cls(sock = s[1], + proc = subprocess.Popen(argv, stdin = s[0], stdout = s[0], close_fds = True), + killsig = signal.SIGINT, + host = host, port = port) + + @classmethod + def tls(cls, host, port): + """ + Set up TLS connection and start listening for first PDU. + + NB: This uses OpenSSL's "s_client" command, which does not + check server certificates properly, so this is not suitable for + production use. Fixing this would be a trivial change, it just + requires using a client program which does check certificates + properly (eg, gnutls-cli, or stunnel's client mode if that works + for such purposes this week). + """ + args = ("openssl", "s_client", "-tls1", "-quiet", "-connect", "%s:%s" % (host, port)) + blather("[Running: %s]" % " ".join(args)) + s = socket.socketpair() + return cls(sock = s[1], + proc = subprocess.Popen(args, stdin = s[0], stdout = s[0], close_fds = True), + killsig = signal.SIGKILL, + host = host, port = port) + + def setup_sql(self, sqlname): + """ + Set up an SQLite database to contain the table we receive. If + necessary, we will create the database. + """ + import sqlite3 + missing = not os.path.exists(sqlname) + self.sql = sqlite3.connect(sqlname, detect_types = sqlite3.PARSE_DECLTYPES) + self.sql.text_factory = str + cur = self.sql.cursor() + cur.execute("PRAGMA foreign_keys = on") + if missing: + cur.execute(''' + CREATE TABLE cache ( + cache_id INTEGER PRIMARY KEY NOT NULL, + host TEXT NOT NULL, + port TEXT NOT NULL, + nonce INTEGER, + serial INTEGER, + updated INTEGER, + UNIQUE (host, port))''') + cur.execute(''' + CREATE TABLE prefix ( + cache_id INTEGER NOT NULL + REFERENCES cache(cache_id) + ON DELETE CASCADE + ON UPDATE CASCADE, + asn INTEGER NOT NULL, + prefix TEXT NOT NULL, + prefixlen INTEGER NOT NULL, + max_prefixlen INTEGER NOT NULL, + UNIQUE (cache_id, asn, prefix, prefixlen, max_prefixlen))''') + + cur.execute(''' + CREATE TABLE routerkey ( + cache_id INTEGER NOT NULL + REFERENCES cache(cache_id) + ON DELETE CASCADE + ON UPDATE CASCADE, + asn INTEGER NOT NULL, + ski TEXT NOT NULL, + key TEXT NOT NULL, + UNIQUE (cache_id, asn, ski), + UNIQUE (cache_id, asn, key))''') + + cur.execute("SELECT cache_id, nonce, serial FROM cache WHERE host = ? AND port = ?", + (self.host, self.port)) + try: + self.cache_id, self.current_nonce, self.current_serial = cur.fetchone() + except TypeError: + cur.execute("INSERT INTO cache (host, port) VALUES (?, ?)", (self.host, self.port)) + self.cache_id = cur.lastrowid + self.sql.commit() + + def cache_reset(self): + """ + Handle cache_reset actions. + """ + self.current_serial = None + if self.sql: + cur = self.sql.cursor() + cur.execute("DELETE FROM prefix WHERE cache_id = ?", (self.cache_id,)) + cur.execute("UPDATE cache SET serial = NULL WHERE cache_id = ?", (self.cache_id,)) + + def end_of_data(self, serial, nonce): + """ + Handle end_of_data actions. + """ + self.current_serial = serial + self.current_nonce = nonce + if self.sql: + self.sql.execute("UPDATE cache SET serial = ?, nonce = ?, updated = datetime('now') WHERE cache_id = ?", + (serial, nonce, self.cache_id)) + self.sql.commit() + + def consume_prefix(self, prefix): + """ + Handle one prefix PDU. + """ + if self.sql: + values = (self.cache_id, prefix.asn, str(prefix.prefix), prefix.prefixlen, prefix.max_prefixlen) + if prefix.announce: + self.sql.execute("INSERT INTO prefix (cache_id, asn, prefix, prefixlen, max_prefixlen) " + "VALUES (?, ?, ?, ?, ?)", + values) + else: + self.sql.execute("DELETE FROM prefix " + "WHERE cache_id = ? AND asn = ? AND prefix = ? AND prefixlen = ? AND max_prefixlen = ?", + values) + + + def consume_routerkey(self, routerkey): + """ + Handle one Router Key PDU. + """ + + if self.sql: + values = (self.cache_id, routerkey.asn, + base64.urlsafe_b64encode(routerkey.ski).rstrip("="), + base64.b64encode(routerkey.key)) + if routerkey.announce: + self.sql.execute("INSERT INTO routerkey (cache_id, asn, ski, key) " + "VALUES (?, ?, ?, ?)", + values) + else: + self.sql.execute("DELETE FROM routerkey " + "WHERE cache_id = ? AND asn = ? AND (ski = ? OR key = ?)", + values) + + + def deliver_pdu(self, pdu): + """ + Handle received PDU. + """ + pdu.consume(self) + + def push_pdu(self, pdu): + """ + Log outbound PDU then write it to stream. + """ + blather(pdu) + pdu_channel.push_pdu(self, pdu) + + def cleanup(self): + """ + Force clean up this client's child process. If everything goes + well, child will have exited already before this method is called, + but we may need to whack it with a stick if something breaks. + """ + if self.proc is not None and self.proc.returncode is None: + try: + os.kill(self.proc.pid, self.killsig) + except OSError: + pass + + def handle_close(self): + """ + Intercept close event so we can log it, then shut down. + """ + blather("Server closed channel") + pdu_channel.handle_close(self) + +class kickme_channel(asyncore.dispatcher): + """ + asyncore dispatcher for the PF_UNIX socket that cronjob mode uses to + kick servers when it's time to send notify PDUs to clients. + """ + + def __init__(self, server): + asyncore.dispatcher.__init__(self) + self.server = server + self.sockname = "%s.%d" % (kickme_base, os.getpid()) + self.create_socket(socket.AF_UNIX, socket.SOCK_DGRAM) + try: + self.bind(self.sockname) + os.chmod(self.sockname, 0660) + except socket.error, e: + log("Couldn't bind() kickme socket: %r" % e) + self.close() + except OSError, e: + log("Couldn't chmod() kickme socket: %r" % e) + + def writable(self): + """ + This socket is read-only, never writable. + """ + return False + + def handle_connect(self): + """ + Ignore connect events (not very useful on datagram socket). + """ + pass + + def handle_read(self): + """ + Handle receipt of a datagram. + """ + data = self.recv(512) + self.server.notify(data) + + def cleanup(self): + """ + Clean up this dispatcher's socket. + """ + self.close() + try: + os.unlink(self.sockname) + except: + pass + + def log(self, msg): + """ + Intercept asyncore's logging. + """ + log(msg) + + def log_info(self, msg, tag = "info"): + """ + Intercept asyncore's logging. + """ + log("asyncore: %s: %s" % (tag, msg)) + + def handle_error(self): + """ + Handle errors caught by asyncore main loop. + """ + c, e = sys.exc_info()[:2] + if backtrace_on_exceptions or e == 0: + for line in traceback.format_exc().splitlines(): + log(line) + else: + log("[Exception: %s: %s]" % (c.__name__, e)) + log("[Exiting after unhandled exception]") + sys.exit(1) + + +def hostport_tag(): + """ + Construct hostname/address + port when we're running under a + protocol we understand well enough to do that. This is all + kludgery. Just grit your teeth, or perhaps just close your eyes. + """ + + proto = None + + if proto is None: + try: + host, port = socket.fromfd(0, socket.AF_INET, socket.SOCK_STREAM).getpeername() + proto = "tcp" + except: + pass + + if proto is None: + try: + host, port = socket.fromfd(0, socket.AF_INET6, socket.SOCK_STREAM).getpeername()[0:2] + proto = "tcp" + except: + pass + + if proto is None: + try: + host, port = os.environ["SSH_CONNECTION"].split()[0:2] + proto = "ssh" + except: + pass + + if proto is None: + try: + host, port = os.environ["REMOTE_HOST"], os.getenv("REMOTE_PORT") + proto = "ssl" + except: + pass + + if proto is None: + return "" + elif not port: + return "/%s/%s" % (proto, host) + elif ":" in host: + return "/%s/%s.%s" % (proto, host, port) + else: + return "/%s/%s:%s" % (proto, host, port) + + +def kick_all(serial): + """ + Kick any existing server processes to wake them up. + """ + + try: + os.stat(kickme_dir) + except OSError: + blather('# Creating directory "%s"' % kickme_dir) + os.makedirs(kickme_dir) + + msg = "Good morning, serial %d is ready" % serial + sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) + for name in glob.iglob("%s.*" % kickme_base): + try: + blather("# Kicking %s" % name) + sock.sendto(msg, name) + except socket.error: + try: + blather("# Failed to kick %s, probably dead socket, attempting cleanup" % name) + os.unlink(name) + except Exception, e: + blather("# Couldn't unlink suspected dead socket %s: %s" % (name, e)) + except Exception, e: + log("# Failed to kick %s and don't understand why: %s" % (name, e)) + sock.close() + +def cronjob_main(argv): + """ + Run this mode right after rcynic to do the real work of groveling + through the ROAs that rcynic collects and translating that data into + the form used in the rpki-router protocol. This mode prepares both + full dumps (AXFR) and incremental dumps against a specific prior + version (IXFR). [Terminology here borrowed from DNS, as is much of + the protocol design.] Finally, this mode kicks any active servers, + so that they can notify their clients that a new version is + available. + + Run this in the directory where you want to write its output files, + which should also be the directory in which you run this program in + --server mode. + + This mode takes one argument on the command line, which specifies + the directory name of rcynic's authenticated output tree (normally + $somewhere/rcynic-data/authenticated/). + """ + + if len(argv) != 1: + sys.exit("Expected one argument, got %r" % (argv,)) + + old_ixfrs = glob.glob("*.ix.*") + + current = read_current()[0] + cutoff = timestamp.now(-(24 * 60 * 60)) + for f in glob.iglob("*.ax"): + t = timestamp(int(f.split(".")[0])) + if t < cutoff and t != current: + blather("# Deleting old file %s, timestamp %s" % (f, t)) + os.unlink(f) + + pdus = axfr_set.parse_rcynic(argv[0]) + if pdus == axfr_set.load_current(): + blather("# No change, new version not needed") + sys.exit() + pdus.save_axfr() + for axfr in glob.iglob("*.ax"): + if axfr != pdus.filename(): + pdus.save_ixfr(axfr_set.load(axfr)) + pdus.mark_current() + + blather("# New serial is %d (%s)" % (pdus.serial, pdus.serial)) + + kick_all(pdus.serial) + + old_ixfrs.sort() + for ixfr in old_ixfrs: + try: + blather("# Deleting old file %s" % ixfr) + os.unlink(ixfr) + except OSError: + pass + +def show_main(argv): + """ + Display dumps created by --cronjob mode in textual form. + Intended only for debugging. + + This mode takes no command line arguments. Run it in the directory + where you ran --cronjob mode. + """ + + if argv: + sys.exit("Unexpected arguments: %r" % (argv,)) + + g = glob.glob("*.ax") + g.sort() + for f in g: + axfr_set.load(f).show() + + g = glob.glob("*.ix.*") + g.sort() + for f in g: + ixfr_set.load(f).show() + +def server_main(argv): + """ + Implement the server side of the rpkk-router protocol. Other than + one PF_UNIX socket inode, this doesn't write anything to disk, so it + can be run with minimal privileges. Most of the hard work has + already been done in --cronjob mode, so all that this mode has to do + is serve up the results. + + In production use this server should run under sshd. The subsystem + mechanism in sshd does not allow us to pass arguments on the command + line, so setting this up might require a wrapper script, but in + production use you will probably want to lock down the public key + used to authenticate the ssh session so that it can only run this + one command, in which case you can just specify the full command + including any arguments in the authorized_keys file. + + Unless you do something special, sshd will have this program running + in whatever it thinks is the home directory associated with the + username given in the ssh prototocol setup, so it may be easiest to + set this up so that the home directory sshd puts this program into + is the one where --cronjob left its files for this mode to pick up. + + This mode must be run in the directory where you ran --cronjob mode. + + This mode takes one optional argument: if provided, the argument is + the name of a directory to which the program should chdir() on + startup; this may simplify setup when running under inetd. + + The server is event driven, so everything interesting happens in the + channel classes. + """ + + blather("[Starting]") + if len(argv) > 1: + sys.exit("Unexpected arguments: %r" % (argv,)) + if argv: + try: + os.chdir(argv[0]) + except OSError, e: + sys.exit(e) + kickme = None + try: + server = server_channel() + kickme = kickme_channel(server = server) + asyncore.loop(timeout = None) + except KeyboardInterrupt: + sys.exit(0) + finally: + if kickme is not None: + kickme.cleanup() + + +def listener_tcp_main(argv): + """ + Simple plain-TCP listener. Listens on a specified TCP port, upon + receiving a connection, forks the process and starts child executing + at server_main(). + + First argument (required) is numeric port number. + + Second argument (optional) is directory, like --server. + + NB: plain-TCP is completely insecure. We only implement this + because it's all that the routers currently support. In theory, we + will all be running TCP-AO in the future, at which point this will + go away. + """ + + # Perhaps we should daemonize? Deal with that later. + + if len(argv) > 2: + sys.exit("Unexpected arguments: %r" % (argv,)) + try: + port = int(argv[0]) if argv[0].isdigit() else socket.getservbyname(argv[0], "tcp") + except: + sys.exit("Couldn't parse port number on which to listen") + if len(argv) > 1: + try: + os.chdir(argv[1]) + except OSError, e: + sys.exit(e) + listener = None + try: + listener = socket.socket(socket.AF_INET6, socket.SOCK_STREAM) + listener.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0) + except: + if listener is not None: + listener.close() + listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + try: + listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) + except AttributeError: + pass + listener.bind(("", port)) + listener.listen(5) + blather("[Listening on port %s]" % port) + while True: + s, ai = listener.accept() + blather("[Received connection from %r]" % (ai,)) + pid = os.fork() + if pid == 0: + os.dup2(s.fileno(), 0) + os.dup2(s.fileno(), 1) + s.close() + #os.closerange(3, os.sysconf("SC_OPEN_MAX")) + global log_tag + log_tag = "rtr-origin/server" + hostport_tag() + syslog.closelog() + syslog.openlog(log_tag, syslog.LOG_PID, syslog_facility) + server_main(()) + sys.exit() + else: + blather("[Spawned server %d]" % pid) + try: + while True: + pid, status = os.waitpid(0, os.WNOHANG) + if pid: + blather("[Server %s exited]" % pid) + else: + break + except: + pass + + +def client_main(argv): + """ + Toy client, intended only for debugging. + + This program takes one or more arguments. The first argument + determines what kind of connection it should open to the server, the + remaining arguments are connection details specific to this + particular type of connection. + + If the first argument is "loopback", the client will run a copy of + the server directly in a subprocess, and communicate with it via a + PF_UNIX socket pair. This sub-mode takes no further arguments. + + If the first argument is "ssh", the client will attempt to run ssh + in as subprocess to connect to the server using the ssh subsystem + mechanism as specified for this protocol. The remaining arguments + should be a hostname (or IP address in a form acceptable to ssh) and + a TCP port number. + + If the first argument is "tcp", the client will attempt to open a + direct (and completely insecure!) TCP connection to the server. + The remaining arguments should be a hostname (or IP address) and + a TCP port number. + + If the first argument is "tls", the client will attempt to open a + TLS connection to the server. The remaining arguments should be a + hostname (or IP address) and a TCP port number. + + An optional final name is the name of a file containing a SQLite + database in which to store the received table. If specified, this + database will be created if missing. + """ + + blather("[Startup]") + client = None + if not argv: + argv = ["loopback"] + proto = argv[0] + if proto == "loopback" and len(argv) in (1, 2): + constructor = client_channel.loopback + host, port = "", "" + sqlname = None if len(argv) == 1 else argv[1] + elif proto in ("ssh", "tcp", "tls") and len(argv) in (3, 4): + constructor = getattr(client_channel, proto) + host, port = argv[1:3] + sqlname = None if len(argv) == 3 else argv[3] + else: + sys.exit("Unexpected arguments: %s" % " ".join(argv)) + + try: + client = constructor(host, port) + if sqlname: + client.setup_sql(sqlname) + while True: + if client.current_serial is None or client.current_nonce is None: + client.push_pdu(reset_query()) + else: + client.push_pdu(serial_query(serial = client.current_serial, nonce = client.current_nonce)) + wakeup = time.time() + 600 + while True: + remaining = wakeup - time.time() + if remaining < 0: + break + asyncore.loop(timeout = remaining, count = 1) + + except KeyboardInterrupt: + sys.exit(0) + finally: + if client is not None: + client.cleanup() + +def bgpdump_convert_main(argv): + """ + Simulate route origin data from a set of BGP dump files. + + * DANGER WILL ROBINSON! * + * DEBUGGING AND TEST USE ONLY! * + + argv is an ordered list of filenames. Each file must be a BGP RIB + dumps, a BGP UPDATE dumps, or an AXFR dump in the format written by + this program's --cronjob command. The first file must be a RIB dump + or AXFR dump, it cannot be an UPDATE dump. Output will be a set of + AXFR and IXFR files with timestamps derived from the BGP dumps, + which can be used as input to this program's --server command for + test purposes. SUCH DATA PROVIDE NO SECURITY AT ALL. + + You have been warned. + """ + + first = True + db = None + axfrs = [] + + for filename in argv: + + if filename.endswith(".ax"): + blather("Reading %s" % filename) + db = axfr_set.load(filename) + + elif os.path.basename(filename).startswith("ribs."): + db = axfr_set.parse_bgpdump_rib_dump(filename) + db.save_axfr() + + elif not first: + assert db is not None + db.parse_bgpdump_update(filename) + db.save_axfr() + + else: + sys.exit("First argument must be a RIB dump or .ax file, don't know what to do with %s" % filename) + + blather("DB serial now %d (%s)" % (db.serial, db.serial)) + if first and read_current() == (None, None): + db.mark_current() + first = False + + for axfr in axfrs: + blather("Loading %s" % axfr) + ax = axfr_set.load(axfr) + blather("Computing changes from %d (%s) to %d (%s)" % (ax.serial, ax.serial, db.serial, db.serial)) + db.save_ixfr(ax) + del ax + + axfrs.append(db.filename()) + + +def bgpdump_select_main(argv): + """ + Simulate route origin data from a set of BGP dump files. + + * DANGER WILL ROBINSON! * + * DEBUGGING AND TEST USE ONLY! * + + Set current serial number to correspond to an .ax file created by + converting BGP dump files. SUCH DATA PROVIDE NO SECURITY AT ALL. + + You have been warned. + """ + + serial = None + try: + head, sep, tail = os.path.basename(argv[0]).partition(".") + if len(argv) == 1 and head.isdigit() and sep == "." and tail == "ax": + serial = timestamp(head) + except: + pass + if serial is None: + sys.exit("Argument must be name of a .ax file") + + nonce = read_current()[1] + if nonce is None: + nonce = new_nonce() + + write_current(serial, nonce) + kick_all(serial) + + +class bgpsec_replay_clock(object): + """ + Internal clock for replaying BGP dump files. + + * DANGER WILL ROBINSON! * + * DEBUGGING AND TEST USE ONLY! * + + This class replaces the normal on-disk serial number mechanism with + an in-memory version based on pre-computed data. + bgpdump_server_main() uses this hack to replay historical data for + testing purposes. DO NOT USE THIS IN PRODUCTION. + + You have been warned. + """ + + def __init__(self): + self.timestamps = [timestamp(int(f.split(".")[0])) for f in glob.iglob("*.ax")] + self.timestamps.sort() + self.offset = self.timestamps[0] - int(time.time()) + self.nonce = new_nonce() + + def __nonzero__(self): + return len(self.timestamps) > 0 + + def now(self): + return timestamp.now(self.offset) + + def read_current(self): + now = self.now() + while len(self.timestamps) > 1 and now >= self.timestamps[1]: + del self.timestamps[0] + return self.timestamps[0], self.nonce + + def siesta(self): + now = self.now() + if len(self.timestamps) <= 1: + return None + elif now < self.timestamps[1]: + return self.timestamps[1] - now + else: + return 1 + + +def bgpdump_server_main(argv): + """ + Simulate route origin data from a set of BGP dump files. + + * DANGER WILL ROBINSON! * + * DEBUGGING AND TEST USE ONLY! * + + This is a clone of server_main() which replaces the external serial + number updates triggered via the kickme channel by cronjob_main with + an internal clocking mechanism to replay historical test data. + + DO NOT USE THIS IN PRODUCTION. + + You have been warned. + """ + + blather("[Starting]") + if len(argv) > 1: + sys.exit("Unexpected arguments: %r" % (argv,)) + if argv: + try: + os.chdir(argv[0]) + except OSError, e: + sys.exit(e) + # + # Yes, this really does replace a global function with a bound + # method to our clock object. Fun stuff, huh? + # + global read_current + clock = bgpsec_replay_clock() + read_current = clock.read_current + # + try: + server = server_channel() + old_serial = server.get_serial() + blather("[Starting at serial %d (%s)]" % (old_serial, old_serial)) + while clock: + new_serial = server.get_serial() + if old_serial != new_serial: + blather("[Serial bumped from %d (%s) to %d (%s)]" % (old_serial, old_serial, new_serial, new_serial)) + server.notify() + old_serial = new_serial + asyncore.loop(timeout = clock.siesta(), count = 1) + except KeyboardInterrupt: + sys.exit(0) + +# Figure out where the scan_roas utility program is today +try: + # Set from autoconf + scan_roas = rpki.autoconf.scan_roas +except NameError: + # Source directory + scan_roas = os.path.normpath(os.path.join(sys.path[0], "..", "utils", + "scan_roas", "scan_roas")) +# If that didn't work, use $PATH and hope for the best +if not os.path.exists(scan_roas): + scan_roas = "scan_roas" + +# Same thing for scan_routercerts +try: + # Set from autoconf + scan_routercerts = rpki.autoconf.scan_routercerts +except NameError: + # Source directory + scan_routercerts = os.path.normpath(os.path.join(sys.path[0], "..", "utils", + "scan_routercerts", "scan_routercerts")) +if not os.path.exists(scan_routercerts): + scan_routercerts = "scan_routercerts" + +force_zero_nonce = False + +kickme_dir = "sockets" +kickme_base = os.path.join(kickme_dir, "kickme") + +main_dispatch = { + "cronjob" : cronjob_main, + "client" : client_main, + "server" : server_main, + "show" : show_main, + "listener_tcp" : listener_tcp_main, + "bgpdump_convert" : bgpdump_convert_main, + "bgpdump_select" : bgpdump_select_main, + "bgpdump_server" : bgpdump_server_main } + +def usage(msg = None): + f = sys.stderr if msg else sys.stdout + f.write("Usage: %s [options] --mode [arguments]\n" % sys.argv[0]) + f.write("\n") + f.write("where options are zero or more of:\n") + f.write("\n") + f.write("--syslog facility.warning_priority[.info_priority]\n") + f.write("\n") + f.write("--zero-nonce\n") + f.write("\n") + f.write("and --mode is one of:\n") + f.write("\n") + for name, func in main_dispatch.iteritems(): + f.write("--%s:\n" % name) + f.write(func.__doc__) + f.write("\n") + sys.exit(msg) + +if __name__ == "__main__": + + os.environ["TZ"] = "UTC" + time.tzset() + + mode = None + + syslog_facility, syslog_warning, syslog_info = syslog.LOG_DAEMON, syslog.LOG_WARNING, syslog.LOG_INFO + + opts, argv = getopt.getopt(sys.argv[1:], "hs:z?", ["help", "syslog=", "zero-nonce"] + main_dispatch.keys()) + for o, a in opts: + if o in ("-h", "--help", "-?"): + usage() + elif o in ("-z", "--zero-nonce"): + force_zero_nonce = True + elif o in ("-s", "--syslog"): + try: + a = [getattr(syslog, "LOG_" + i.upper()) for i in a.split(".")] + if len(a) == 2: + a.append(a[1]) + syslog_facility, syslog_warning, syslog_info = a + if syslog_facility < 8 or syslog_warning >= 8 or syslog_info >= 8: + raise ValueError + except: + usage("Bad value specified for --syslog option") + elif len(o) > 2 and o[2:] in main_dispatch: + if mode is not None: + sys.exit("Conflicting modes specified") + mode = o[2:] + + if mode is None: + usage("No mode specified") + + log_tag = "rtr-origin/" + mode + + if mode in ("server", "bgpdump_server"): + log_tag += hostport_tag() + + if mode in ("cronjob", "server" , "bgpdump_server"): + syslog.openlog(log_tag, syslog.LOG_PID, syslog_facility) + def log(msg): + return syslog.syslog(syslog_warning, str(msg)) + def blather(msg): + return syslog.syslog(syslog_info, str(msg)) + + elif mode == "show": + def log(msg): + try: + os.write(sys.stdout.fileno(), "%s\n" % msg) + except OSError, e: + if e.errno != errno.EPIPE: + raise + blather = log + + else: + def log(msg): + sys.stderr.write("%s %s[%d]: %s\n" % (time.strftime("%F %T"), log_tag, os.getpid(), msg)) + blather = log + + main_dispatch[mode](argv) diff --git a/rp/rpki-rtr/rules.darwin.mk b/rp/rpki-rtr/rules.darwin.mk new file mode 100644 index 00000000..1230db92 --- /dev/null +++ b/rp/rpki-rtr/rules.darwin.mk @@ -0,0 +1,9 @@ +# $Id$ + +install-always: install-binary + +install-postconf: install-listener + +install-listener: + @echo "No rule for $@ on this platform (yet), you'll have to do that yourself if it matters." + diff --git a/rp/rpki-rtr/rules.freebsd.mk b/rp/rpki-rtr/rules.freebsd.mk new file mode 100644 index 00000000..df99da47 --- /dev/null +++ b/rp/rpki-rtr/rules.freebsd.mk @@ -0,0 +1,37 @@ +# $Id$ + +install-always: install-binary + +install-postconf: install-listener + +install-listener: .FORCE + @if /usr/bin/egrep -q '^rpki-rtr' /etc/services ; \ + then \ + echo "You already have a /etc/services entry for rpki-rtr, so I will use it."; \ + elif echo >>/etc/services "rpki-rtr ${RPKI_RTR_PORT}/tcp #RFC 6810" ; \ + then \ + echo "Added rpki-rtr to /etc/services."; \ + else \ + echo "Adding rpki-rtr to /etc/services failed, please fix this, then try again."; \ + exit 1; \ + fi + @if /usr/bin/egrep -q "rpki-rtr[ ]+stream[ ]+tcp[ ]" /etc/inetd.conf; \ + then \ + echo "You already have an inetd.conf entry for rpki-rtr on TCPv4, so I will use it."; \ + elif echo >>/etc/inetd.conf "rpki-rtr stream tcp nowait rpkirtr /usr/local/bin/rtr-origin rtr-origin --server /var/rcynic/rpki-rtr"; \ + then \ + echo "Added rpki-rtr for TCPv4 to /etc/inetd.conf."; \ + else \ + echo "Adding rpki-rtr for TCPv4 to /etc/inetd.conf failed, please fix this, then try again."; \ + exit 1; \ + fi + @if /usr/bin/egrep -q "rpki-rtr[ ]+stream[ ]+tcp6[ ]" /etc/inetd.conf; \ + then \ + echo "You already have an inetd.conf entry for rpki-rtr on TCPv6, so I will use it."; \ + elif echo >>/etc/inetd.conf "rpki-rtr stream tcp6 nowait rpkirtr /usr/local/bin/rtr-origin rtr-origin --server /var/rcynic/rpki-rtr"; \ + then \ + echo "Added rpki-rtr for TCPv6 to /etc/inetd.conf."; \ + else \ + echo "Adding rpki-rtr for TCPv6 to /etc/inetd.conf failed, please fix this, then try again."; \ + exit 1; \ + fi diff --git a/rp/rpki-rtr/rules.linux.mk b/rp/rpki-rtr/rules.linux.mk new file mode 100644 index 00000000..23f90f69 --- /dev/null +++ b/rp/rpki-rtr/rules.linux.mk @@ -0,0 +1,29 @@ +# $Id$ + +install-always: install-binary install-listener + +install-postconf: + @true + +# Only need to make listener if not already present + +install-listener: ${DESTDIR}/etc/xinetd.d/rpki-rtr + +${DESTDIR}/etc/xinetd.d/rpki-rtr: + @${AWK} 'BEGIN { \ + print "service rpki-rtr"; \ + print "{"; \ + print " type = UNLISTED"; \ + print " flags = IPv4"; \ + print " socket_type = stream"; \ + print " protocol = tcp"; \ + print " port = ${RPKI_RTR_PORT}"; \ + print " wait = no"; \ + print " user = rpkirtr"; \ + print " server = ${bindir}/${BIN}"; \ + print " server_args = --server /var/rcynic/rpki-rtr"; \ + print "}"; \ + }' >xinetd.rpki-rtr + ${INSTALL} -d ${DESTDIR}/etc/xinetd.d + ${INSTALL} -m 644 xinetd.rpki-rtr $@ + rm xinetd.rpki-rtr diff --git a/rp/rpki-rtr/rules.unknown.mk b/rp/rpki-rtr/rules.unknown.mk new file mode 100644 index 00000000..fb16e93a --- /dev/null +++ b/rp/rpki-rtr/rules.unknown.mk @@ -0,0 +1,8 @@ +# $Id$ + +install-always: install-binary + +install-postconf: install-listener + +install-listener: + @echo "Don't know how to make $@ on this platform"; exit 1 diff --git a/rp/rpki-rtr/server.sh b/rp/rpki-rtr/server.sh new file mode 100755 index 00000000..7ccf2f38 --- /dev/null +++ b/rp/rpki-rtr/server.sh @@ -0,0 +1,17 @@ +#!/bin/sh - +# +# Wrapper for rtr-origin.py in server mode, for testing. +# +# In production we would probably want to handle all of this either +# directly in the Python code or in the command= setting for a +# particular ssh key, but for initial testing it's simpler to run a +# shall script to change to the right directory and supply any +# necessary command line arguments. +# +# Be warned that almost any error here will cause the subsystem to +# fail mysteriously, leaving behind naught but a SIGCHILD log message +# from sshd as this script dies. + +cd /u/sra/rpki/subvert-rpki.hactrn.net/rtr-origin + +exec /usr/local/bin/python rtr-origin.py --server diff --git a/rp/rpki-rtr/sshd.conf b/rp/rpki-rtr/sshd.conf new file mode 100644 index 00000000..df32ca67 --- /dev/null +++ b/rp/rpki-rtr/sshd.conf @@ -0,0 +1,23 @@ +# $Id$ +# +# sshd config file for testing. Invoke thusly: +# +# /usr/sbin/sshd -f $top/rpki-rtr/sshd.conf -d + +Port 2222 +Protocol 2 +ListenAddress 127.0.0.1 +ListenAddress ::1 +HostKey /u/sra/rpki/subvert-rpki.hactrn.net/rpki-rtr/ssh_host_rsa_key +PermitRootLogin no +PubkeyAuthentication yes +AuthorizedKeysFile /u/sra/rpki/subvert-rpki.hactrn.net/rpki-rtr/authorized_keys +PasswordAuthentication no +PermitEmptyPasswords no +ChallengeResponseAuthentication no +UsePAM no +AllowTcpForwarding no +X11Forwarding no +UseDNS no +PidFile /u/sra/rpki/subvert-rpki.hactrn.net/rpki-rtr/sshd.pid +Subsystem rpki-rtr /u/sra/rpki/subvert-rpki.hactrn.net/rpki-rtr/server.sh |