diff options
Diffstat (limited to 'rpki/left_right.py')
| -rw-r--r-- | rpki/left_right.py | 1071 |
1 files changed, 486 insertions, 585 deletions
diff --git a/rpki/left_right.py b/rpki/left_right.py index c8b6d19b..7189f888 100644 --- a/rpki/left_right.py +++ b/rpki/left_right.py @@ -21,12 +21,13 @@ RPKI "left-right" protocol. """ +import base64 import logging -import rpki.resource_set +import collections + import rpki.x509 import rpki.sql import rpki.exceptions -import rpki.xml_utils import rpki.http import rpki.up_down import rpki.relaxng @@ -36,51 +37,203 @@ import rpki.publication import rpki.async import rpki.rpkid_tasks -logger = logging.getLogger(__name__) - -## @var enforce_strict_up_down_xml_sender -# Enforce strict checking of XML "sender" field in up-down protocol +from lxml.etree import Element, SubElement, tostring as ElementToString -enforce_strict_up_down_xml_sender = False - -class left_right_namespace(object): - """ - XML namespace parameters for left-right protocol. - """ +logger = logging.getLogger(__name__) - xmlns = rpki.relaxng.left_right.xmlns - nsmap = rpki.relaxng.left_right.nsmap -class data_elt(rpki.xml_utils.data_elt, rpki.sql.sql_persistent, left_right_namespace): +xmlns = rpki.relaxng.left_right.xmlns +nsmap = rpki.relaxng.left_right.nsmap +version = rpki.relaxng.left_right.version + +tag_bpki_cert = xmlns + "bpki_cert" +tag_bpki_glue = xmlns + "bpki_glue" +tag_bsc = xmlns + "bsc" +tag_child = xmlns + "child" +tag_list_ee_certificate_requests = xmlns + "list_ee_certificate_requests" +tag_list_ghostbuster_requests = xmlns + "list_ghostbuster_requests" +tag_list_published_objects = xmlns + "list_published_objects" +tag_list_received_resources = xmlns + "list_received_resources" +tag_list_resources = xmlns + "list_resources" +tag_list_roa_requests = xmlns + "list_roa_requests" +tag_msg = xmlns + "msg" +tag_parent = xmlns + "parent" +tag_pkcs10 = xmlns + "pkcs10" +tag_pkcs10_request = xmlns + "pkcs10_request" +tag_report_error = xmlns + "report_error" +tag_repository = xmlns + "repository" +tag_self = xmlns + "self" +tag_signing_cert = xmlns + "signing_cert" +tag_signing_cert_crl = xmlns + "signing_cert_crl" + + +class base_elt(rpki.sql.sql_persistent): """ - Virtual class for top-level left-right protocol data elements. + Virtual class for persistent left-right protocol elements. + These classes are being phased out in favor of Django ORM models. """ handles = () + attributes = () + elements = () + booleans = () self_id = None self_handle = None + def __str__(self): + return ElementToString(self.toXML(), pretty_print = True, encoding = "us-ascii") + + @classmethod + def fromXML(cls, elt): + self = cls() + for key in self.attributes: + val = elt.get(key, None) + if val is not None: + val = val.encode("ascii") + if isinstance(self.attributes, dict) and self.attributes[key] is not None: + val = self.attributes[key](val) + elif val.isdigit() and not key.endswith("_handle"): + val = long(val) + setattr(self, key, val) + for key in self.booleans: + setattr(self, key, elt.get(key, False)) + for b64 in elt: + assert b64.tag.startswith(xmlns) + setattr(self, b64.tag[len(xmlns):], self.elements[b64.tag](Base64 = b64.text)) + return self + + def toXML(self): + elt = Element(self.element_name, nsmap = nsmap) + for key in self.attributes: + val = getattr(self, key, None) + if val is not None: + elt.set(key, str(val)) + for key in self.booleans: + if getattr(self, key, False): + elt.set(key, "yes") + for name in self.elements: + value = getattr(self, name[len(xmlns):], None) + if value is not None and not value.empty(): + SubElement(elt, name, nsmap = nsmap).text = value.get_Base64() + return elt + + def make_reply(self, r_pdu = None): + if r_pdu is None: + r_pdu = self.__class__() + self.make_reply_clone_hook(r_pdu) + handle_name = self.element_name[len(xmlns):] + "_handle" + setattr(r_pdu, handle_name, getattr(self, handle_name, None)) + else: + self.make_reply_clone_hook(r_pdu) + for b in r_pdu.booleans: + setattr(r_pdu, b, False) + r_pdu.action = self.action + r_pdu.tag = self.tag + return r_pdu + + def serve_fetch_one(self): + """ + Find the object on which a get, set, or destroy method should + operate. + """ + + r = self.serve_fetch_one_maybe() + if r is None: + raise rpki.exceptions.NotFound + return r + + def serve_post_save_hook(self, q_pdu, r_pdu, cb, eb): + cb() + + def serve_create(self, r_msg, cb, eb): + r_pdu = self.make_reply() + def one(): + self.sql_store() + setattr(r_pdu, self.sql_template.index, getattr(self, self.sql_template.index)) + self.serve_post_save_hook(self, r_pdu, two, eb) + def two(): + r_msg.append(r_pdu) + cb() + oops = self.serve_fetch_one_maybe() + if oops is not None: + raise rpki.exceptions.DuplicateObject("Object already exists: %r[%r] %r[%r]" % ( + self, getattr(self, self.element_name[len(xmlns):] + "_handle"), + oops, getattr(oops, oops.element_name[len(xmlns):] + "_handle"))) + self.serve_pre_save_hook(self, r_pdu, one, eb) + + def serve_set(self, r_msg, cb, eb): + db_pdu = self.serve_fetch_one() + r_pdu = self.make_reply() + for a in db_pdu.sql_template.columns[1:]: + v = getattr(self, a, None) + if v is not None: + setattr(db_pdu, a, v) + db_pdu.sql_mark_dirty() + def one(): + db_pdu.sql_store() + db_pdu.serve_post_save_hook(self, r_pdu, two, eb) + def two(): + r_msg.append(r_pdu) + cb() + db_pdu.serve_pre_save_hook(self, r_pdu, one, eb) + + def serve_get(self, r_msg, cb, eb): + r_pdu = self.serve_fetch_one() + self.make_reply(r_pdu) + r_msg.append(r_pdu) + cb() + + def serve_list(self, r_msg, cb, eb): + for r_pdu in self.serve_fetch_all(): + self.make_reply(r_pdu) + r_msg.append(r_pdu) + cb() + + def serve_destroy_hook(self, cb, eb): + cb() + + def serve_destroy(self, r_msg, cb, eb): + def done(): + db_pdu.sql_delete() + r_msg.append(self.make_reply()) + cb() + db_pdu = self.serve_fetch_one() + db_pdu.serve_destroy_hook(done, eb) + + def serve_dispatch(self, r_msg, cb, eb): + # Transition hack: handle the .toXML() call for old handlers. + fake_r_msg = [] + def fake_convert(): + r_msg.extend(r_pdu.toXML() if isinstance(r_pdu, base_elt) else r_pdu + for r_pdu in fake_r_msg) + def fake_cb(): + fake_convert() + cb() + def fake_eb(e): + fake_convert() + eb(e) + method = getattr(self, "serve_" + self.action, None) + if method is None: + raise rpki.exceptions.BadQuery("Unexpected query: action %s" % self.action) + method(fake_r_msg, fake_cb, fake_eb) + + def unimplemented_control(self, *controls): + unimplemented = [x for x in controls if getattr(self, x, False)] + if unimplemented: + raise rpki.exceptions.NotImplementedYet("Unimplemented control %s" % ", ".join(unimplemented)) + @property @rpki.sql.cache_reference def self(self): - """ - Fetch self object to which this object links. - """ return self_elt.sql_fetch(self.gctx, self.self_id) @property @rpki.sql.cache_reference def bsc(self): - """ - Return BSC object to which this object links. - """ return bsc_elt.sql_fetch(self.gctx, self.bsc_id) def make_reply_clone_hook(self, r_pdu): - """ - Set handles when cloning, including _id -> _handle translation. - """ if r_pdu.self_handle is None: r_pdu.self_handle = self.self_handle for tag, elt in self.handles: @@ -94,36 +247,23 @@ class data_elt(rpki.xml_utils.data_elt, rpki.sql.sql_persistent, left_right_name @classmethod def serve_fetch_handle(cls, gctx, self_id, handle): - """ - Find an object based on its handle. - """ - return cls.sql_fetch_where1(gctx, cls.element_name + "_handle = %s AND self_id = %s", (handle, self_id)) + name = cls.element_name[len(xmlns):] + return cls.sql_fetch_where1(gctx, name + "_handle = %s AND self_id = %s", (handle, self_id)) def serve_fetch_one_maybe(self): - """ - Find the object on which a get, set, or destroy method should - operate, or which would conflict with a create method. - """ - where = "%s.%s_handle = %%s AND %s.self_id = self.self_id AND self.self_handle = %%s" % ((self.element_name,) * 3) - args = (getattr(self, self.element_name + "_handle"), self.self_handle) + name = self.element_name[len(xmlns):] + where = "%s.%s_handle = %%s AND %s.self_id = self.self_id AND self.self_handle = %%s" % (name, name, name) + args = (getattr(self, name + "_handle"), self.self_handle) + logger.debug(".serve_fetch_one_maybe() %s %s", args[0], args[1]) return self.sql_fetch_where1(self.gctx, where, args, "self") def serve_fetch_all(self): - """ - Find the objects on which a list method should operate. - """ - where = "%s.self_id = self.self_id and self.self_handle = %%s" % self.element_name + name = self.element_name[len(xmlns):] + where = "%s.self_id = self.self_id and self.self_handle = %%s" % name return self.sql_fetch_where(self.gctx, where, (self.self_handle,), "self") def serve_pre_save_hook(self, q_pdu, r_pdu, cb, eb): - """ - Hook to do _handle => _id translation before saving. - - self is always the object to be saved to SQL. For create - operations, self and q_pdu are be the same object; for set - operations, self is the pre-existing object from SQL and q_pdu is - the set request received from the the IRBE. - """ + # self is always the object to be saved to SQL. for tag, elt in self.handles: id_name = tag + "_id" if getattr(self, id_name, None) is None: @@ -133,17 +273,21 @@ class data_elt(rpki.xml_utils.data_elt, rpki.sql.sql_persistent, left_right_name setattr(self, id_name, getattr(x, id_name)) cb() -class self_elt(data_elt): + +class self_elt(base_elt): """ <self/> element. """ - element_name = "self" + element_name = xmlns + "self" attributes = ("action", "tag", "self_handle", "crl_interval", "regen_margin") - elements = ("bpki_cert", "bpki_glue") booleans = ("rekey", "reissue", "revoke", "run_now", "publish_world_now", "revoke_forgotten", "clear_replay_protection") + elements = collections.OrderedDict(( + (tag_bpki_cert, rpki.x509.X509), + (tag_bpki_glue, rpki.x509.X509))) + sql_template = rpki.sql.template( "self", "self_id", @@ -168,58 +312,33 @@ class self_elt(data_elt): @property def bscs(self): - """ - Fetch all BSC objects that link to this self object. - """ return bsc_elt.sql_fetch_where(self.gctx, "self_id = %s", (self.self_id,)) @property def repositories(self): - """ - Fetch all repository objects that link to this self object. - """ return repository_elt.sql_fetch_where(self.gctx, "self_id = %s", (self.self_id,)) @property def parents(self): - """ - Fetch all parent objects that link to this self object. - """ return parent_elt.sql_fetch_where(self.gctx, "self_id = %s", (self.self_id,)) @property def children(self): - """ - Fetch all child objects that link to this self object. - """ return child_elt.sql_fetch_where(self.gctx, "self_id = %s", (self.self_id,)) @property def roas(self): - """ - Fetch all ROA objects that link to this self object. - """ return rpki.rpkid.roa_obj.sql_fetch_where(self.gctx, "self_id = %s", (self.self_id,)) @property def ghostbusters(self): - """ - Fetch all Ghostbuster record objects that link to this self object. - """ return rpki.rpkid.ghostbuster_obj.sql_fetch_where(self.gctx, "self_id = %s", (self.self_id,)) @property def ee_certificates(self): - """ - Fetch all EE certificate objects that link to this self object. - """ return rpki.rpkid.ee_cert_obj.sql_fetch_where(self.gctx, "self_id = %s", (self.self_id,)) - def serve_post_save_hook(self, q_pdu, r_pdu, cb, eb): - """ - Extra server actions for self_elt. - """ actions = [] if q_pdu.rekey: actions.append(self.serve_rekey) @@ -240,96 +359,101 @@ class self_elt(data_elt): rpki.async.iterator(actions, loop, cb) def serve_rekey(self, cb, eb): - """ - Handle a left-right rekey action for this self. - """ def loop(iterator, parent): parent.serve_rekey(iterator, eb) rpki.async.iterator(self.parents, loop, cb) def serve_revoke(self, cb, eb): - """ - Handle a left-right revoke action for this self. - """ def loop(iterator, parent): parent.serve_revoke(iterator, eb) rpki.async.iterator(self.parents, loop, cb) def serve_reissue(self, cb, eb): - """ - Handle a left-right reissue action for this self. - """ def loop(iterator, parent): parent.serve_reissue(iterator, eb) rpki.async.iterator(self.parents, loop, cb) def serve_revoke_forgotten(self, cb, eb): - """ - Handle a left-right revoke_forgotten action for this self. - """ def loop(iterator, parent): parent.serve_revoke_forgotten(iterator, eb) rpki.async.iterator(self.parents, loop, cb) def serve_clear_replay_protection(self, cb, eb): - """ - Handle a left-right clear_replay_protection action for this self. - """ def loop(iterator, obj): obj.serve_clear_replay_protection(iterator, eb) rpki.async.iterator(self.parents + self.children + self.repositories, loop, cb) def serve_destroy_hook(self, cb, eb): - """ - Extra cleanup actions when destroying a self_elt. - """ def loop(iterator, parent): parent.delete(iterator) rpki.async.iterator(self.parents, loop, cb) def serve_publish_world_now(self, cb, eb): - """ - Handle a left-right publish_world_now action for this self. - - The publication stuff needs refactoring, right now publication is - interleaved with local operations in a way that forces far too - many bounces through the task system for any complex update. The - whole thing ought to be rewritten to queue up outgoing publication - PDUs and only send them when we're all done or when we need to - force publication at a particular point in a multi-phase operation. - - Once that reorganization has been done, this method should be - rewritten to reuse the low-level publish() methods that each - object will have...but we're not there yet. So, for now, we just - do this via brute force. Think of it as a trial version to see - whether we've identified everything that needs to be republished - for this operation. - """ + publisher = rpki.rpkid.publication_queue() + repositories = set() + objects = dict() def loop(iterator, parent): - q_msg = rpki.publication.msg.query() - for ca in parent.cas: - ca_detail = ca.active_ca_detail - if ca_detail is not None: - q_msg.append(rpki.publication.crl_elt.make_publish( - ca_detail.crl_uri, ca_detail.latest_crl)) - q_msg.append(rpki.publication.manifest_elt.make_publish( - ca_detail.manifest_uri, ca_detail.latest_manifest)) - q_msg.extend(rpki.publication.certificate_elt.make_publish( - c.uri, c.cert) for c in ca_detail.child_certs) - q_msg.extend(rpki.publication.roa_elt.make_publish( - r.uri, r.roa) for r in ca_detail.roas if r.roa is not None) - q_msg.extend(rpki.publication.ghostbuster_elt.make_publish( - g.uri, g.ghostbuster) for g in ca_detail.ghostbusters) - parent.repository.call_pubd(iterator, eb, q_msg) + repository = parent.repository + if repository.peer_contact_uri in repositories: + return iterator() + repositories.add(repository.peer_contact_uri) + q_msg = Element(rpki.publication.tag_msg, nsmap = rpki.publication.nsmap, + type = "query", version = rpki.publication.version) + SubElement(q_msg, rpki.publication.tag_list, tag = "list") + def list_handler(r_pdu): + rpki.publication.raise_if_error(r_pdu) + assert r_pdu.tag == rpki.publication.tag_list + assert r_pdu.get("uri") not in objects + objects[r_pdu.get("uri")] = (r_pdu.get("hash"), repository) + repository.call_pubd(iterator, eb, + q_msg, length_check = False, + handlers = dict(list = list_handler)) + + def reconcile(uri, obj, repository): + h, r = objects.pop(uri, (None, None)) + if h is not None: + assert r == repository + publisher.queue(uri = uri, new_obj = obj, old_hash = h, repository = repository) - rpki.async.iterator(self.parents, loop, cb) + def done(): + for parent in self.parents: + repository = parent.repository + for ca in parent.cas: + ca_detail = ca.active_ca_detail + if ca_detail is not None: + reconcile(uri = ca_detail.crl_uri, + obj = ca_detail.latest_crl, + repository = repository) + reconcile(uri = ca_detail.manifest_uri, + obj = ca_detail.latest_manifest, + repository = repository) + for c in ca_detail.child_certs: + reconcile(uri = c.uri, + obj = c.cert, + repository = repository) + for r in ca_detail.roas: + if r.roa is not None: + reconcile(uri = r.uri, + obj = r.roa, + repository = repository) + for g in ca_detail.ghostbusters: + reconcile(uri = g.uri, + obj = g.ghostbuster, + repository = repository) + for c in ca_detail.ee_certificates: + reconcile(uri = c.uri, + obj = c.cert, + repository = repository) + for u in objects: + h, r = objects[u] + publisher.queue(uri = u, old_hash = h, repository = r) + publisher.call_pubd(cb, eb) + + rpki.async.iterator(self.parents, loop, done) def serve_run_now(self, cb, eb): - """ - Handle a left-right run_now action for this self. - """ logger.debug("Forced immediate run of periodic actions for self %s[%d]", self.self_handle, self.self_id) completion = rpki.rpkid_tasks.CompletionHandler(cb) @@ -342,13 +466,11 @@ class self_elt(data_elt): Find the self object upon which a get, set, or destroy action should operate, or which would conflict with a create method. """ + return self.serve_fetch_handle(self.gctx, None, self.self_handle) @classmethod def serve_fetch_handle(cls, gctx, self_id, self_handle): - """ - Find a self object based on its self_handle. - """ return cls.sql_fetch_where1(gctx, "self_handle = %s", (self_handle,)) def serve_fetch_all(self): @@ -357,16 +479,12 @@ class self_elt(data_elt): This is different from the list action for all other objects, where list only works within a given self_id context. """ + return self.sql_fetch_all(self.gctx) def schedule_cron_tasks(self, completion): - """ - Schedule periodic tasks. - """ - if self.cron_tasks is None: self.cron_tasks = tuple(task(self) for task in rpki.rpkid_tasks.task_classes) - for task in self.cron_tasks: self.gctx.task_add(task) completion.register(task) @@ -392,16 +510,20 @@ class self_elt(data_elt): return results -class bsc_elt(data_elt): +class bsc_elt(base_elt): """ <bsc/> (Business Signing Context) element. """ - element_name = "bsc" + element_name = xmlns + "bsc" attributes = ("action", "tag", "self_handle", "bsc_handle", "key_type", "hash_alg", "key_length") - elements = ("signing_cert", "signing_cert_crl", "pkcs10_request") booleans = ("generate_keypair",) + elements = collections.OrderedDict(( + (tag_signing_cert, rpki.x509.X509), + (tag_signing_cert_crl, rpki.x509.CRL), + (tag_pkcs10_request, rpki.x509.PKCS10))) + sql_template = rpki.sql.template( "bsc", "bsc_id", @@ -425,47 +547,42 @@ class bsc_elt(data_elt): @property def repositories(self): - """ - Fetch all repository objects that link to this BSC object. - """ return repository_elt.sql_fetch_where(self.gctx, "bsc_id = %s", (self.bsc_id,)) @property def parents(self): - """ - Fetch all parent objects that link to this BSC object. - """ return parent_elt.sql_fetch_where(self.gctx, "bsc_id = %s", (self.bsc_id,)) @property def children(self): - """ - Fetch all child objects that link to this BSC object. - """ return child_elt.sql_fetch_where(self.gctx, "bsc_id = %s", (self.bsc_id,)) def serve_pre_save_hook(self, q_pdu, r_pdu, cb, eb): """ - Extra server actions for bsc_elt -- handle key generation. For - now this only allows RSA with SHA-256. + Extra server actions -- handle key generation, only RSA with SHA-256 for now. """ + if q_pdu.generate_keypair: assert q_pdu.key_type in (None, "rsa") and q_pdu.hash_alg in (None, "sha256") self.private_key_id = rpki.x509.RSA.generate(keylength = q_pdu.key_length or 2048) self.pkcs10_request = rpki.x509.PKCS10.create(keypair = self.private_key_id) r_pdu.pkcs10_request = self.pkcs10_request - data_elt.serve_pre_save_hook(self, q_pdu, r_pdu, cb, eb) + super(bsc_elt, self).serve_pre_save_hook(q_pdu, r_pdu, cb, eb) -class repository_elt(data_elt): + +class repository_elt(base_elt): """ <repository/> element. """ - element_name = "repository" - attributes = ("action", "tag", "self_handle", "repository_handle", "bsc_handle", "peer_contact_uri") - elements = ("bpki_cert", "bpki_glue") + element_name = xmlns + "repository" + attributes = ("action", "tag", "self_handle", "repository_handle", "bsc_handle", "peer_contact_uri", "rrdp_notification_uri") booleans = ("clear_replay_protection",) + elements = collections.OrderedDict(( + (tag_bpki_cert, rpki.x509.X509), + (tag_bpki_glue, rpki.x509.X509))) + sql_template = rpki.sql.template( "repository", "repository_id", @@ -473,6 +590,7 @@ class repository_elt(data_elt): "self_id", "bsc_id", "peer_contact_uri", + "rrdp_notification_uri", ("bpki_cert", rpki.x509.X509), ("bpki_glue", rpki.x509.X509), ("last_cms_timestamp", rpki.sundial.datetime)) @@ -483,21 +601,16 @@ class repository_elt(data_elt): bpki_cert = None bpki_glue = None last_cms_timestamp = None + rrdp_notification_uri = None def __repr__(self): return rpki.log.log_repr(self, self.repository_handle) @property def parents(self): - """ - Fetch all parent objects that link to this repository object. - """ return parent_elt.sql_fetch_where(self.gctx, "repository_id = %s", (self.repository_id,)) def serve_post_save_hook(self, q_pdu, r_pdu, cb, eb): - """ - Extra server actions for repository_elt. - """ actions = [] if q_pdu.clear_replay_protection: actions.append(self.serve_clear_replay_protection) @@ -506,45 +619,33 @@ class repository_elt(data_elt): rpki.async.iterator(actions, loop, cb) def serve_clear_replay_protection(self, cb, eb): - """ - Handle a left-right clear_replay_protection action for this repository. - """ self.last_cms_timestamp = None self.sql_mark_dirty() cb() - @staticmethod - def default_pubd_handler(pdu): - """ - Default handler for publication response PDUs. - """ - pdu.raise_if_error() - def call_pubd(self, callback, errback, q_msg, handlers = None): + def call_pubd(self, callback, errback, q_msg, handlers = {}, length_check = True): # pylint: disable=W0102 """ Send a message to publication daemon and return the response. As a convenience, attempting to send an empty message returns immediate success without sending anything. - Handlers is a dict of handler functions to process the response + handlers is a dict of handler functions to process the response PDUs. If the tag value in the response PDU appears in the dict, the associated handler is called to process the PDU. If no tag - matches, default_pubd_handler() is called. A handler value of - False suppresses calling of the default handler. + matches, a default handler is called to check for errors; a + handler value of False suppresses calling of the default handler. """ try: self.gctx.sql.sweep() - if not q_msg: + if len(q_msg) == 0: return callback() - if handlers is None: - handlers = {} - for q_pdu in q_msg: - logger.info("Sending %s %s to pubd", q_pdu.action, q_pdu.uri) + logger.info("Sending %r to pubd", q_pdu) bsc = self.bsc q_der = rpki.publication.cms_msg().wrap(q_msg, bsc.private_key_id, bsc.signing_cert, bsc.signing_cert_crl) @@ -557,11 +658,11 @@ class repository_elt(data_elt): r_msg = r_cms.unwrap(bpki_ta_path) r_cms.check_replay_sql(self, self.peer_contact_uri) for r_pdu in r_msg: - handler = handlers.get(r_pdu.tag, self.default_pubd_handler) + handler = handlers.get(r_pdu.get("tag"), rpki.publication.raise_if_error) if handler: logger.debug("Calling pubd handler %r", handler) handler(r_pdu) - if len(q_msg) != len(r_msg): + if length_check and len(q_msg) != len(r_msg): raise rpki.exceptions.BadPublicationReply("Wrong number of response PDUs from pubd: sent %r, got %r" % (q_msg, r_msg)) callback() except (rpki.async.ExitNow, SystemExit): @@ -581,17 +682,21 @@ class repository_elt(data_elt): except Exception, e: errback(e) -class parent_elt(data_elt): + +class parent_elt(base_elt): """ <parent/> element. """ - element_name = "parent" + element_name = xmlns + "parent" attributes = ("action", "tag", "self_handle", "parent_handle", "bsc_handle", "repository_handle", "peer_contact_uri", "sia_base", "sender_name", "recipient_name") - elements = ("bpki_cms_cert", "bpki_cms_glue") booleans = ("rekey", "reissue", "revoke", "revoke_forgotten", "clear_replay_protection") + elements = collections.OrderedDict(( + (tag_bpki_cert, rpki.x509.X509), + (tag_bpki_glue, rpki.x509.X509))) + sql_template = rpki.sql.template( "parent", "parent_id", @@ -603,16 +708,16 @@ class parent_elt(data_elt): "sia_base", "sender_name", "recipient_name", - ("bpki_cms_cert", rpki.x509.X509), - ("bpki_cms_glue", rpki.x509.X509), + ("bpki_cert", rpki.x509.X509), + ("bpki_glue", rpki.x509.X509), ("last_cms_timestamp", rpki.sundial.datetime)) handles = (("self", self_elt), ("bsc", bsc_elt), ("repository", repository_elt)) - bpki_cms_cert = None - bpki_cms_glue = None + bpki_cert = None + bpki_glue = None last_cms_timestamp = None def __repr__(self): @@ -621,22 +726,13 @@ class parent_elt(data_elt): @property @rpki.sql.cache_reference def repository(self): - """ - Fetch repository object to which this parent object links. - """ return repository_elt.sql_fetch(self.gctx, self.repository_id) @property def cas(self): - """ - Fetch all CA objects that link to this parent object. - """ return rpki.rpkid.ca_obj.sql_fetch_where(self.gctx, "parent_id = %s", (self.parent_id,)) def serve_post_save_hook(self, q_pdu, r_pdu, cb, eb): - """ - Extra server actions for parent_elt. - """ actions = [] if q_pdu.rekey: actions.append(self.serve_rekey) @@ -653,33 +749,21 @@ class parent_elt(data_elt): rpki.async.iterator(actions, loop, cb) def serve_rekey(self, cb, eb): - """ - Handle a left-right rekey action for this parent. - """ def loop(iterator, ca): ca.rekey(iterator, eb) rpki.async.iterator(self.cas, loop, cb) def serve_revoke(self, cb, eb): - """ - Handle a left-right revoke action for this parent. - """ def loop(iterator, ca): ca.revoke(cb = iterator, eb = eb) rpki.async.iterator(self.cas, loop, cb) def serve_reissue(self, cb, eb): - """ - Handle a left-right reissue action for this parent. - """ def loop(iterator, ca): ca.reissue(cb = iterator, eb = eb) rpki.async.iterator(self.cas, loop, cb) def serve_clear_replay_protection(self, cb, eb): - """ - Handle a left-right clear_replay_protection action for this parent. - """ self.last_cms_timestamp = None self.sql_mark_dirty() cb() @@ -696,10 +780,11 @@ class parent_elt(data_elt): """ def done(r_msg): - cb(dict((rc.class_name, set(c.cert.gSKI() for c in rc.certs)) - for rc in r_msg.payload.classes)) - - rpki.up_down.list_pdu.query(self, done, eb) + cb(dict((rc.get("class_name"), + set(rpki.x509.X509(Base64 = c.text).gSKI() + for c in rc.getiterator(rpki.up_down.tag_certificate))) + for rc in r_msg.getiterator(rpki.up_down.tag_class))) + self.up_down_list_query(done, eb) def revoke_skis(self, rc_name, skis_to_revoke, cb, eb): @@ -708,12 +793,10 @@ class parent_elt(data_elt): """ def loop(iterator, ski): + def revoked(r_pdu): + iterator() logger.debug("Asking parent %r to revoke class %r, SKI %s", self, rc_name, ski) - q_pdu = rpki.up_down.revoke_pdu() - q_pdu.class_name = rc_name - q_pdu.ski = ski - self.query_up_down(q_pdu, lambda r_pdu: iterator(), eb) - + self.up_down_revoke_query(rc_name, ski, revoked, eb) rpki.async.iterator(skis_to_revoke, loop, cb) @@ -775,17 +858,39 @@ class parent_elt(data_elt): def serve_destroy_hook(self, cb, eb): - """ - Extra server actions when destroying a parent_elt. - """ - self.delete(cb, delete_parent = False) - def query_up_down(self, q_pdu, cb, eb): - """ - Client code for sending one up-down query PDU to this parent. - """ + def _compose_up_down_query(self, query_type): + return Element(rpki.up_down.tag_message, nsmap = rpki.up_down.nsmap, version = rpki.up_down.version, + sender = self.sender_name, recipient = self.recipient_name, type = query_type) + + + def up_down_list_query(self, cb, eb): + q_msg = self._compose_up_down_query("list") + self.query_up_down(q_msg, cb, eb) + + + def up_down_issue_query(self, ca, ca_detail, cb, eb): + pkcs10 = rpki.x509.PKCS10.create( + keypair = ca_detail.private_key_id, + is_ca = True, + caRepository = ca.sia_uri, + rpkiManifest = ca_detail.manifest_uri, + rpkiNotify = ca.parent.repository.rrdp_notification_uri) + q_msg = self._compose_up_down_query("issue") + q_pdu = SubElement(q_msg, rpki.up_down.tag_request, class_name = ca.parent_resource_class) + q_pdu.text = pkcs10.get_Base64() + self.query_up_down(q_msg, cb, eb) + + + def up_down_revoke_query(self, class_name, ski, cb, eb): + q_msg = self._compose_up_down_query("revoke") + SubElement(q_msg, rpki.up_down.tag_key, class_name = class_name, ski = ski) + self.query_up_down(q_msg, cb, eb) + + + def query_up_down(self, q_msg, cb, eb): bsc = self.bsc if bsc is None: @@ -794,11 +899,6 @@ class parent_elt(data_elt): if bsc.signing_cert is None: raise rpki.exceptions.BSCNotReady("BSC %r[%s] is not yet usable" % (bsc.bsc_handle, bsc.bsc_id)) - q_msg = rpki.up_down.message_pdu.make_query( - payload = q_pdu, - sender = self.sender_name, - recipient = self.recipient_name) - q_der = rpki.up_down.cms_msg().wrap(q_msg, bsc.private_key_id, bsc.signing_cert, bsc.signing_cert_crl) @@ -809,10 +909,11 @@ class parent_elt(data_elt): r_msg = r_cms.unwrap((self.gctx.bpki_ta, self.self.bpki_cert, self.self.bpki_glue, - self.bpki_cms_cert, - self.bpki_cms_glue)) + self.bpki_cert, + self.bpki_glue)) r_cms.check_replay_sql(self, self.peer_contact_uri) - r_msg.payload.check_response() + rpki.up_down.check_response(r_msg, q_msg.get("type")) + except (SystemExit, rpki.async.ExitNow): raise except Exception, e: @@ -827,16 +928,20 @@ class parent_elt(data_elt): errback = eb, content_type = rpki.up_down.content_type) -class child_elt(data_elt): + +class child_elt(base_elt): """ <child/> element. """ - element_name = "child" + element_name = xmlns + "child" attributes = ("action", "tag", "self_handle", "child_handle", "bsc_handle") - elements = ("bpki_cert", "bpki_glue") booleans = ("reissue", "clear_replay_protection") + elements = collections.OrderedDict(( + (tag_bpki_cert, rpki.x509.X509), + (tag_bpki_glue, rpki.x509.X509))) + sql_template = rpki.sql.template( "child", "child_id", @@ -858,29 +963,17 @@ class child_elt(data_elt): return rpki.log.log_repr(self, self.child_handle) def fetch_child_certs(self, ca_detail = None, ski = None, unique = False): - """ - Fetch all child_cert objects that link to this child object. - """ return rpki.rpkid.child_cert_obj.fetch(self.gctx, self, ca_detail, ski, unique) @property def child_certs(self): - """ - Fetch all child_cert objects that link to this child object. - """ return self.fetch_child_certs() @property def parents(self): - """ - Fetch all parent objects that link to self object to which this child object links. - """ return parent_elt.sql_fetch_where(self.gctx, "self_id = %s", (self.self_id,)) def serve_post_save_hook(self, q_pdu, r_pdu, cb, eb): - """ - Extra server actions for child_elt. - """ actions = [] if q_pdu.reissue: actions.append(self.serve_reissue) @@ -891,26 +984,17 @@ class child_elt(data_elt): rpki.async.iterator(actions, loop, cb) def serve_reissue(self, cb, eb): - """ - Handle a left-right reissue action for this child. - """ publisher = rpki.rpkid.publication_queue() for child_cert in self.child_certs: child_cert.reissue(child_cert.ca_detail, publisher, force = True) publisher.call_pubd(cb, eb) def serve_clear_replay_protection(self, cb, eb): - """ - Handle a left-right clear_replay_protection action for this child. - """ self.last_cms_timestamp = None self.sql_mark_dirty() cb() def ca_from_class_name(self, class_name): - """ - Fetch the CA corresponding to an up-down class_name. - """ if not class_name.isdigit(): raise rpki.exceptions.BadClassNameSyntax("Bad class name %s" % class_name) ca = rpki.rpkid.ca_obj.sql_fetch(self.gctx, long(class_name)) @@ -924,368 +1008,185 @@ class child_elt(data_elt): return ca def serve_destroy_hook(self, cb, eb): - """ - Extra server actions when destroying a child_elt. - """ publisher = rpki.rpkid.publication_queue() for child_cert in self.child_certs: child_cert.revoke(publisher = publisher, generate_crl_and_manifest = True) publisher.call_pubd(cb, eb) - def serve_up_down(self, query, callback): - """ - Outer layer of server handling for one up-down PDU from this child. - """ - - bsc = self.bsc - if bsc is None: - raise rpki.exceptions.BSCNotFound("Could not find BSC %s" % self.bsc_id) - q_cms = rpki.up_down.cms_msg(DER = query) - q_msg = q_cms.unwrap((self.gctx.bpki_ta, - self.self.bpki_cert, - self.self.bpki_glue, - self.bpki_cert, - self.bpki_glue)) - q_cms.check_replay_sql(self, "child", self.child_handle) - q_msg.payload.gctx = self.gctx - if enforce_strict_up_down_xml_sender and q_msg.sender != self.child_handle: - raise rpki.exceptions.BadSender("Unexpected XML sender %s" % q_msg.sender) - self.gctx.sql.sweep() - - def done(r_msg): - # - # Exceptions from this point on are problematic, as we have no - # sane way of reporting errors in the error reporting mechanism. - # May require refactoring, ignore the issue for now. - # - reply = rpki.up_down.cms_msg().wrap(r_msg, bsc.private_key_id, - bsc.signing_cert, bsc.signing_cert_crl) - callback(reply) - - try: - q_msg.serve_top_level(self, done) - except (rpki.async.ExitNow, SystemExit): - raise - except rpki.exceptions.NoActiveCA, data: - done(q_msg.serve_error(data)) - except Exception, e: - logger.exception("Unhandled exception serving up-down request from %r", self) - done(q_msg.serve_error(e)) - -class list_resources_elt(rpki.xml_utils.base_elt, left_right_namespace): - """ - <list_resources/> element. - """ - - element_name = "list_resources" - attributes = ("self_handle", "tag", "child_handle", "valid_until", "asn", "ipv4", "ipv6") - valid_until = None - - def __repr__(self): - return rpki.log.log_repr(self, self.self_handle, self.child_handle, self.asn, self.ipv4, self.ipv6) - - def startElement(self, stack, name, attrs): - """ - Handle <list_resources/> element. This requires special handling - due to the data types of some of the attributes. - """ - assert name == "list_resources", "Unexpected name %s, stack %s" % (name, stack) - self.read_attrs(attrs) - if isinstance(self.valid_until, str): - self.valid_until = rpki.sundial.datetime.fromXMLtime(self.valid_until) - if self.asn is not None: - self.asn = rpki.resource_set.resource_set_as(self.asn) - if self.ipv4 is not None: - self.ipv4 = rpki.resource_set.resource_set_ipv4(self.ipv4) - if self.ipv6 is not None: - self.ipv6 = rpki.resource_set.resource_set_ipv6(self.ipv6) - - def toXML(self): - """ - Generate <list_resources/> element. This requires special - handling due to the data types of some of the attributes. - """ - elt = self.make_elt() - if isinstance(self.valid_until, int): - elt.set("valid_until", self.valid_until.toXMLtime()) - return elt - -class list_roa_requests_elt(rpki.xml_utils.base_elt, left_right_namespace): - """ - <list_roa_requests/> element. - """ - - element_name = "list_roa_requests" - attributes = ("self_handle", "tag", "asn", "ipv4", "ipv6") - - def startElement(self, stack, name, attrs): - """ - Handle <list_roa_requests/> element. This requires special handling - due to the data types of some of the attributes. - """ - assert name == "list_roa_requests", "Unexpected name %s, stack %s" % (name, stack) - self.read_attrs(attrs) - if self.ipv4 is not None: - self.ipv4 = rpki.resource_set.roa_prefix_set_ipv4(self.ipv4) - if self.ipv6 is not None: - self.ipv6 = rpki.resource_set.roa_prefix_set_ipv6(self.ipv6) - - def __repr__(self): - return rpki.log.log_repr(self, self.self_handle, self.asn, self.ipv4, self.ipv6) - -class list_ghostbuster_requests_elt(rpki.xml_utils.text_elt, left_right_namespace): - """ - <list_ghostbuster_requests/> element. - """ - - element_name = "list_ghostbuster_requests" - attributes = ("self_handle", "tag", "parent_handle") - text_attribute = "vcard" - - vcard = None - - def __repr__(self): - return rpki.log.log_repr(self, self.self_handle, self.parent_handle) - -class list_ee_certificate_requests_elt(rpki.xml_utils.base_elt, left_right_namespace): - """ - <list_ee_certificate_requests/> element. - """ - - element_name = "list_ee_certificate_requests" - attributes = ("self_handle", "tag", "gski", "valid_until", "asn", "ipv4", "ipv6", "cn", "sn", "eku") - elements = ("pkcs10",) - pkcs10 = None - valid_until = None - eku = None + def up_down_handle_list(self, q_msg, r_msg, callback, errback): - def __repr__(self): - return rpki.log.log_repr(self, self.self_handle, self.gski, self.cn, self.sn, self.asn, self.ipv4, self.ipv6) + def got_resources(irdb_resources): - def startElement(self, stack, name, attrs): - """ - Handle <list_ee_certificate_requests/> element. This requires special - handling due to the data types of some of the attributes. - """ - if name not in self.elements: - assert name == self.element_name, "Unexpected name %s, stack %s" % (name, stack) - self.read_attrs(attrs) - if isinstance(self.valid_until, str): - self.valid_until = rpki.sundial.datetime.fromXMLtime(self.valid_until) - if self.asn is not None: - self.asn = rpki.resource_set.resource_set_as(self.asn) - if self.ipv4 is not None: - self.ipv4 = rpki.resource_set.resource_set_ipv4(self.ipv4) - if self.ipv6 is not None: - self.ipv6 = rpki.resource_set.resource_set_ipv6(self.ipv6) - if self.eku is not None: - self.eku = self.eku.split(",") - - def endElement(self, stack, name, text): - """ - Handle <pkcs10/> sub-element. - """ - assert len(self.elements) == 1 - if name == self.elements[0]: - self.pkcs10 = rpki.x509.PKCS10(Base64 = text) - else: - assert name == self.element_name, "Unexpected name %s, stack %s" % (name, stack) - stack.pop() + if irdb_resources.valid_until < rpki.sundial.now(): + logger.debug("Child %s's resources expired %s", self.child_handle, irdb_resources.valid_until) + else: + for parent in self.parents: + for ca in parent.cas: + ca_detail = ca.active_ca_detail + if not ca_detail: + logger.debug("No active ca_detail, can't issue to %s", self.child_handle) + continue + resources = ca_detail.latest_ca_cert.get_3779resources() & irdb_resources + if resources.empty(): + logger.debug("No overlap between received resources and what child %s should get ([%s], [%s])", + self.child_handle, ca_detail.latest_ca_cert.get_3779resources(), irdb_resources) + continue + rc = SubElement(r_msg, rpki.up_down.tag_class, + class_name = str(ca.ca_id), + cert_url = ca_detail.ca_cert_uri, + resource_set_as = str(resources.asn), + resource_set_ipv4 = str(resources.v4), + resource_set_ipv6 = str(resources.v6), + resource_set_notafter = str(resources.valid_until)) + for child_cert in self.fetch_child_certs(ca_detail = ca_detail): + c = SubElement(rc, rpki.up_down.tag_certificate, cert_url = child_cert.uri) + c.text = child_cert.cert.get_Base64() + SubElement(rc, rpki.up_down.tag_issuer).text = ca_detail.latest_ca_cert.get_Base64() + callback() + + self.gctx.irdb_query_child_resources(self.self.self_handle, self.child_handle, got_resources, errback) + + + def up_down_handle_issue(self, q_msg, r_msg, callback, errback): + + def got_resources(irdb_resources): + + def done(): + rc = SubElement(r_msg, rpki.up_down.tag_class, + class_name = class_name, + cert_url = ca_detail.ca_cert_uri, + resource_set_as = str(resources.asn), + resource_set_ipv4 = str(resources.v4), + resource_set_ipv6 = str(resources.v6), + resource_set_notafter = str(resources.valid_until)) + c = SubElement(rc, rpki.up_down.tag_certificate, cert_url = child_cert.uri) + c.text = child_cert.cert.get_Base64() + SubElement(rc, rpki.up_down.tag_issuer).text = ca_detail.latest_ca_cert.get_Base64() + callback() + + if irdb_resources.valid_until < rpki.sundial.now(): + raise rpki.exceptions.IRDBExpired("IRDB entry for child %s expired %s" % ( + self.child_handle, irdb_resources.valid_until)) + + resources = irdb_resources & ca_detail.latest_ca_cert.get_3779resources() + resources.valid_until = irdb_resources.valid_until + req_key = pkcs10.getPublicKey() + req_sia = pkcs10.get_SIA() + child_cert = self.fetch_child_certs(ca_detail = ca_detail, ski = req_key.get_SKI(), unique = True) + + # Generate new cert or regenerate old one if necessary + + publisher = rpki.rpkid.publication_queue() + + if child_cert is None: + child_cert = ca_detail.issue( + ca = ca, + child = self, + subject_key = req_key, + sia = req_sia, + resources = resources, + publisher = publisher) + else: + child_cert = child_cert.reissue( + ca_detail = ca_detail, + sia = req_sia, + resources = resources, + publisher = publisher) - def toXML(self): - """ - Generate <list_ee_certificate_requests/> element. This requires special - handling due to the data types of some of the attributes. - """ - if isinstance(self.eku, (tuple, list)): - self.eku = ",".join(self.eku) - elt = self.make_elt() - for i in self.elements: - self.make_b64elt(elt, i, getattr(self, i, None)) - if isinstance(self.valid_until, int): - elt.set("valid_until", self.valid_until.toXMLtime()) - return elt + self.gctx.sql.sweep() + assert child_cert and child_cert.sql_in_db + publisher.call_pubd(done, errback) -class list_published_objects_elt(rpki.xml_utils.text_elt, left_right_namespace): - """ - <list_published_objects/> element. - """ + req = q_msg[0] + assert req.tag == rpki.up_down.tag_request - element_name = "list_published_objects" - attributes = ("self_handle", "tag", "uri", "child_handle") - text_attribute = "obj" + # Subsetting not yet implemented, this is the one place where we + # have to handle it, by reporting that we're lame. - obj = None - child_handle = None + if any(req.get(a) for a in ("req_resource_set_as", "req_resource_set_ipv4", "req_resource_set_ipv6")): + raise rpki.exceptions.NotImplementedYet("req_* attributes not implemented yet, sorry") - def __repr__(self): - return rpki.log.log_repr(self, self.self_handle, self.child_handle, self.uri) + class_name = req.get("class_name") + pkcs10 = rpki.x509.PKCS10(Base64 = req.text) + pkcs10.check_valid_request_ca() + ca = self.ca_from_class_name(class_name) + ca_detail = ca.active_ca_detail + if ca_detail is None: + raise rpki.exceptions.NoActiveCA("No active CA for class %r" % class_name) - def serve_dispatch(self, r_msg, cb, eb): - """ - Handle a <list_published_objects/> query. The method name is a - misnomer here, there's no action attribute and no dispatch, we - just dump every published object for the specified <self/> and return. - """ - for parent in self_elt.serve_fetch_handle(self.gctx, None, self.self_handle).parents: - for ca in parent.cas: - ca_detail = ca.active_ca_detail - if ca_detail is not None: - r_msg.append(self.make_reply(ca_detail.crl_uri, ca_detail.latest_crl)) - r_msg.append(self.make_reply(ca_detail.manifest_uri, ca_detail.latest_manifest)) - r_msg.extend(self.make_reply(c.uri, c.cert, c.child.child_handle) - for c in ca_detail.child_certs) - r_msg.extend(self.make_reply(r.uri, r.roa) - for r in ca_detail.roas if r.roa is not None) - r_msg.extend(self.make_reply(g.uri, g.ghostbuster) - for g in ca_detail.ghostbusters) - r_msg.extend(self.make_reply(c.uri, c.cert) - for c in ca_detail.ee_certificates) - cb() - - def make_reply(self, uri, obj, child_handle = None): - """ - Generate one reply PDU. - """ - r_pdu = self.make_pdu(tag = self.tag, self_handle = self.self_handle, - uri = uri, child_handle = child_handle) - r_pdu.obj = obj.get_Base64() - return r_pdu + self.gctx.irdb_query_child_resources(self.self.self_handle, self.child_handle, got_resources, errback) -class list_received_resources_elt(rpki.xml_utils.base_elt, left_right_namespace): - """ - <list_received_resources/> element. - """ - element_name = "list_received_resources" - attributes = ("self_handle", "tag", "parent_handle", - "notBefore", "notAfter", "uri", "sia_uri", "aia_uri", "asn", "ipv4", "ipv6") - - def __repr__(self): - return rpki.log.log_repr(self, self.self_handle, self.parent_handle, self.uri, self.notAfter) + def up_down_handle_revoke(self, q_msg, r_msg, callback, errback): - def serve_dispatch(self, r_msg, cb, eb): - """ - Handle a <list_received_resources/> query. The method name is a - misnomer here, there's no action attribute and no dispatch, we - just dump a bunch of data about every certificate issued to us by - one of our parents, then return. - """ - for parent in self_elt.serve_fetch_handle(self.gctx, None, self.self_handle).parents: - for ca in parent.cas: - ca_detail = ca.active_ca_detail - if ca_detail is not None and ca_detail.latest_ca_cert is not None: - r_msg.append(self.make_reply(parent.parent_handle, ca_detail.ca_cert_uri, ca_detail.latest_ca_cert)) - cb() - - def make_reply(self, parent_handle, uri, cert): - """ - Generate one reply PDU. - """ - resources = cert.get_3779resources() - return self.make_pdu( - tag = self.tag, - self_handle = self.self_handle, - parent_handle = parent_handle, - notBefore = str(cert.getNotBefore()), - notAfter = str(cert.getNotAfter()), - uri = uri, - sia_uri = cert.get_sia_directory_uri(), - aia_uri = cert.get_aia_uri(), - asn = resources.asn, - ipv4 = resources.v4, - ipv6 = resources.v6) - -class report_error_elt(rpki.xml_utils.text_elt, left_right_namespace): - """ - <report_error/> element. - """ + def done(): + SubElement(r_msg, key.tag, class_name = class_name, ski = key.get("ski")) + callback() - element_name = "report_error" - attributes = ("tag", "self_handle", "error_code") - text_attribute = "error_text" + key = q_msg[0] + assert key.tag == rpki.up_down.tag_key + class_name = key.get("class_name") + ski = base64.urlsafe_b64decode(key.get("ski") + "=") - error_text = None + publisher = rpki.rpkid.publication_queue() - def __repr__(self): - return rpki.log.log_repr(self, self.self_handle, self.error_code) + ca = self.ca_from_class_name(class_name) + for ca_detail in ca.ca_details: + for child_cert in self.fetch_child_certs(ca_detail = ca_detail, ski = ski): + child_cert.revoke(publisher = publisher) - @classmethod - def from_exception(cls, e, self_handle = None, tag = None): - """ - Generate a <report_error/> element from an exception. - """ - self = cls() - self.self_handle = self_handle - self.tag = tag - self.error_code = e.__class__.__name__ - self.error_text = str(e) - return self + self.gctx.sql.sweep() + publisher.call_pubd(done, errback) -class msg(rpki.xml_utils.msg, left_right_namespace): - """ - Left-right PDU. - """ - ## @var version - # Protocol version - version = int(rpki.relaxng.left_right.version) - - ## @var pdus - # Dispatch table of PDUs for this protocol. - pdus = dict((x.element_name, x) - for x in (self_elt, child_elt, parent_elt, bsc_elt, - repository_elt, list_resources_elt, - list_roa_requests_elt, list_ghostbuster_requests_elt, - list_ee_certificate_requests_elt, - list_published_objects_elt, - list_received_resources_elt, report_error_elt)) - - def serve_top_level(self, gctx, cb): + def serve_up_down(self, q_der, callback): """ - Serve one msg PDU. + Outer layer of server handling for one up-down PDU from this child. """ - r_msg = self.__class__.reply() - - def loop(iterator, q_pdu): - - def fail(e): - if not isinstance(e, rpki.exceptions.NotFound): - logger.exception("Unhandled exception serving left-right PDU %r", q_pdu) - r_msg.append(report_error_elt.from_exception( - e, self_handle = q_pdu.self_handle, tag = q_pdu.tag)) - cb(r_msg) + def done(): + callback(rpki.up_down.cms_msg().wrap(r_msg, bsc.private_key_id, + bsc.signing_cert, bsc.signing_cert_crl)) - try: - q_pdu.gctx = gctx - q_pdu.serve_dispatch(r_msg, iterator, fail) - except (rpki.async.ExitNow, SystemExit): - raise - except Exception, e: - fail(e) + def lose(e): + logger.exception("Unhandled exception serving child %r", self) + rpki.up_down.generate_error_response_from_exception(r_msg, e, q_type) + done() - def done(): - cb(r_msg) + bsc = self.bsc + if bsc is None: + raise rpki.exceptions.BSCNotFound("Could not find BSC %s" % self.bsc_id) + q_cms = rpki.up_down.cms_msg(DER = q_der) + q_msg = q_cms.unwrap((self.gctx.bpki_ta, + self.self.bpki_cert, + self.self.bpki_glue, + self.bpki_cert, + self.bpki_glue)) + q_cms.check_replay_sql(self, "child", self.child_handle) + q_type = q_msg.get("type") + logger.info("Serving %s query from child %s [sender %s, recipient %s]", + q_type, self.child_handle, q_msg.get("sender"), q_msg.get("recipient")) + if rpki.up_down.enforce_strict_up_down_xml_sender and q_msg.get("sender") != self.child_handle: + raise rpki.exceptions.BadSender("Unexpected XML sender %s" % q_msg.get("sender")) + self.gctx.sql.sweep() - rpki.async.iterator(self, loop, done) + r_msg = Element(rpki.up_down.tag_message, nsmap = rpki.up_down.nsmap, version = rpki.up_down.version, + sender = q_msg.get("recipient"), recipient = q_msg.get("sender"), type = q_type + "_response") -class sax_handler(rpki.xml_utils.sax_handler): - """ - SAX handler for Left-Right protocol. - """ + try: + getattr(self, "up_down_handle_" + q_type)(q_msg, r_msg, done, lose) + except (rpki.async.ExitNow, SystemExit): + raise + except Exception, e: + lose(e) - pdu = msg - name = "msg" - version = rpki.relaxng.left_right.version class cms_msg(rpki.x509.XML_CMS_object): """ - Class to hold a CMS-signed left-right PDU. + CMS-signed left-right PDU. """ encoding = "us-ascii" schema = rpki.relaxng.left_right - saxify = sax_handler.saxify |