aboutsummaryrefslogtreecommitdiff
path: root/rpki/rpkid.py
diff options
context:
space:
mode:
Diffstat (limited to 'rpki/rpkid.py')
-rw-r--r--rpki/rpkid.py438
1 files changed, 269 insertions, 169 deletions
diff --git a/rpki/rpkid.py b/rpki/rpkid.py
index 36ee2ea9..07a81f79 100644
--- a/rpki/rpkid.py
+++ b/rpki/rpkid.py
@@ -42,8 +42,11 @@ import rpki.async
import rpki.daemonize
import rpki.rpkid_tasks
+from lxml.etree import Element, SubElement
+
logger = logging.getLogger(__name__)
+
class main(object):
"""
Main program for rpkid.
@@ -75,7 +78,7 @@ class main(object):
rpki.log.init("rpkid", args)
- self.cfg = rpki.config.parser(args.config, "rpkid")
+ self.cfg = rpki.config.parser(set_filename = args.config, section = "rpkid")
self.cfg.set_global_flags()
if not args.foreground:
@@ -116,17 +119,6 @@ class main(object):
self.publication_kludge_base = self.cfg.get("publication-kludge-base", "publication/")
- # Icky hack to let Iain do some testing quickly, should go away
- # once we sort out whether we can make this change permanent.
- #
- # OK, the stuff to add router certificate support makes enough
- # other changes that we're going to need a migration program in
- # any case, so might as well throw the switch here too, or at
- # least find out if it (still) works as expected.
-
- self.merge_publication_directories = self.cfg.getboolean("merge_publication_directories",
- True)
-
self.use_internal_cron = self.cfg.getboolean("use-internal-cron", True)
self.initial_delay = random.randint(self.cfg.getint("initial-delay-min", 10),
@@ -283,14 +275,14 @@ class main(object):
up_down_url_regexp = re.compile("/up-down/([-A-Z0-9_]+)/([-A-Z0-9_]+)$", re.I)
- def up_down_handler(self, query, path, cb):
+ def up_down_handler(self, q_der, path, cb):
"""
Process one up-down PDU.
"""
- def done(reply):
+ def done(r_der):
self.sql.sweep()
- cb(200, body = reply)
+ cb(200, body = r_der)
try:
match = self.up_down_url_regexp.search(path)
@@ -303,7 +295,7 @@ class main(object):
"self")
if child is None:
raise rpki.exceptions.ChildNotFound("Could not find child %s of self %s in up_down_handler()" % (child_handle, self_handle))
- child.serve_up_down(query, done)
+ child.serve_up_down(q_der, done)
except (rpki.async.ExitNow, SystemExit):
raise
except (rpki.exceptions.ChildNotFound, rpki.exceptions.BadContactURL), e:
@@ -318,6 +310,7 @@ class main(object):
Record that we were still alive when we got here, by resetting
keepalive timer.
"""
+
if force or self.cron_timeout is not None:
self.cron_timeout = rpki.sundial.now() + self.cron_keepalive
@@ -325,6 +318,7 @@ class main(object):
"""
Add a task to the scheduler task queue, unless it's already queued.
"""
+
if task not in self.task_queue:
logger.debug("Adding %r to task queue", task)
self.task_queue.append(task)
@@ -339,6 +333,7 @@ class main(object):
queue (we don't want to run it directly, as that could eventually
blow out our call stack).
"""
+
try:
self.task_current = self.task_queue.pop(0)
except IndexError:
@@ -350,6 +345,7 @@ class main(object):
"""
Run first task on the task queue, unless one is running already.
"""
+
if self.task_current is None:
self.task_next()
@@ -446,6 +442,7 @@ class ca_obj(rpki.sql.sql_persistent):
"""
Fetch parent object to which this CA object links.
"""
+
return rpki.left_right.parent_elt.sql_fetch(self.gctx, self.parent_id)
@property
@@ -453,6 +450,7 @@ class ca_obj(rpki.sql.sql_persistent):
"""
Fetch all ca_detail objects that link to this CA object.
"""
+
return ca_detail_obj.sql_fetch_where(self.gctx, "ca_id = %s", (self.ca_id,))
@property
@@ -460,6 +458,7 @@ class ca_obj(rpki.sql.sql_persistent):
"""
Fetch the pending ca_details for this CA, if any.
"""
+
return ca_detail_obj.sql_fetch_where(self.gctx, "ca_id = %s AND state = 'pending'", (self.ca_id,))
@property
@@ -467,6 +466,7 @@ class ca_obj(rpki.sql.sql_persistent):
"""
Fetch the active ca_detail for this CA, if any.
"""
+
return ca_detail_obj.sql_fetch_where1(self.gctx, "ca_id = %s AND state = 'active'", (self.ca_id,))
@property
@@ -474,6 +474,7 @@ class ca_obj(rpki.sql.sql_persistent):
"""
Fetch deprecated ca_details for this CA, if any.
"""
+
return ca_detail_obj.sql_fetch_where(self.gctx, "ca_id = %s AND state = 'deprecated'", (self.ca_id,))
@property
@@ -481,6 +482,7 @@ class ca_obj(rpki.sql.sql_persistent):
"""
Fetch active and deprecated ca_details for this CA, if any.
"""
+
return ca_detail_obj.sql_fetch_where(self.gctx, "ca_id = %s AND (state = 'active' OR state = 'deprecated')", (self.ca_id,))
@property
@@ -488,6 +490,7 @@ class ca_obj(rpki.sql.sql_persistent):
"""
Fetch revoked ca_details for this CA, if any.
"""
+
return ca_detail_obj.sql_fetch_where(self.gctx, "ca_id = %s AND state = 'revoked'", (self.ca_id,))
@property
@@ -496,7 +499,7 @@ class ca_obj(rpki.sql.sql_persistent):
Fetch ca_details which are candidates for consideration when
processing an up-down issue_response PDU.
"""
- #return ca_detail_obj.sql_fetch_where(self.gctx, "ca_id = %s AND latest_ca_cert IS NOT NULL AND state != 'revoked'", (self.ca_id,))
+
return ca_detail_obj.sql_fetch_where(self.gctx, "ca_id = %s AND state != 'revoked'", (self.ca_id,))
def construct_sia_uri(self, parent, rc):
@@ -505,16 +508,12 @@ class ca_obj(rpki.sql.sql_persistent):
information and the parent's up-down protocol list_response PDU.
"""
- sia_uri = rc.suggested_sia_head and rc.suggested_sia_head.rsync()
- if not sia_uri or not sia_uri.startswith(parent.sia_base):
+ sia_uri = rc.get("suggested_sia_head", "")
+ if not sia_uri.startswith("rsync://") or not sia_uri.startswith(parent.sia_base):
sia_uri = parent.sia_base
if not sia_uri.endswith("/"):
raise rpki.exceptions.BadURISyntax("SIA URI must end with a slash: %s" % sia_uri)
- # With luck this can go away sometime soon.
- if self.gctx.merge_publication_directories:
- return sia_uri
- else:
- return sia_uri + str(self.ca_id) + "/"
+ return sia_uri
def check_for_updates(self, parent, rc, cb, eb):
"""
@@ -531,28 +530,40 @@ class ca_obj(rpki.sql.sql_persistent):
self.sia_uri = sia_uri
self.sql_mark_dirty()
- rc_resources = rc.to_resource_bag()
- cert_map = dict((c.cert.get_SKI(), c) for c in rc.certs)
+ class_name = rc.get("class_name")
+
+ rc_resources = rpki.resource_set.resource_bag(
+ rc.get("resource_set_as"),
+ rc.get("resource_set_ipv4"),
+ rc.get("resource_set_ipv6"),
+ rc.get("resource_set_notafter"))
+
+ cert_map = {}
+ for c in rc.getiterator(rpki.up_down.tag_certificate):
+ x = rpki.x509.X509(Base64 = c.text)
+ u = rpki.up_down.multi_uri(c.get("cert_url")).rsync()
+ cert_map[x.gSKI()] = (x, u)
def loop(iterator, ca_detail):
self.gctx.checkpoint()
- rc_cert = cert_map.pop(ca_detail.public_key.get_SKI(), None)
+ rc_cert, rc_cert_uri = cert_map.pop(ca_detail.public_key.gSKI(), (None, None))
if rc_cert is None:
- logger.warning("SKI %s in resource class %s is in database but missing from list_response to %s from %s, maybe parent certificate went away?",
- ca_detail.public_key.gSKI(), rc.class_name, parent.self.self_handle, parent.parent_handle)
+ logger.warning("SKI %s in resource class %s is in database but missing from list_response to %s from %s, "
+ "maybe parent certificate went away?",
+ ca_detail.public_key.gSKI(), class_name, parent.self.self_handle, parent.parent_handle)
publisher = publication_queue()
ca_detail.delete(ca = ca_detail.ca, publisher = publisher)
return publisher.call_pubd(iterator, eb)
else:
- if ca_detail.state == "active" and ca_detail.ca_cert_uri != rc_cert.cert_url.rsync():
- logger.debug("AIA changed: was %s now %s", ca_detail.ca_cert_uri, rc_cert.cert_url.rsync())
- ca_detail.ca_cert_uri = rc_cert.cert_url.rsync()
+ if ca_detail.state == "active" and ca_detail.ca_cert_uri != rc_cert_uri:
+ logger.debug("AIA changed: was %s now %s", ca_detail.ca_cert_uri, rc_cert_uri)
+ ca_detail.ca_cert_uri = rc_cert_uri
ca_detail.sql_mark_dirty()
if ca_detail.state in ("pending", "active"):
@@ -564,7 +575,7 @@ class ca_obj(rpki.sql.sql_persistent):
if (ca_detail.state == "pending" or
sia_uri_changed or
- ca_detail.latest_ca_cert != rc_cert.cert or
+ ca_detail.latest_ca_cert != rc_cert or
ca_detail.latest_ca_cert.getNotAfter() != rc_resources.valid_until or
current_resources.undersized(rc_resources) or
current_resources.oversized(rc_resources)):
@@ -582,9 +593,7 @@ class ca_obj(rpki.sql.sql_persistent):
def done():
if cert_map:
logger.warning("Unknown certificate SKI%s %s in resource class %s in list_response to %s from %s, maybe you want to \"revoke_forgotten\"?",
- "" if len(cert_map) == 1 else "s",
- ", ".join(c.cert.gSKI() for c in cert_map.values()),
- rc.class_name, parent.self.self_handle, parent.parent_handle)
+ "" if len(cert_map) == 1 else "s", ", ".join(cert_map), class_name, parent.self.self_handle, parent.parent_handle)
self.gctx.sql.sweep()
self.gctx.checkpoint()
cb()
@@ -592,26 +601,25 @@ class ca_obj(rpki.sql.sql_persistent):
ca_details = self.issue_response_candidate_ca_details
if True:
- skis_parent = set(x.cert.gSKI()
- for x in cert_map.itervalues())
+ skis_parent = set(cert_map)
skis_me = set(x.latest_ca_cert.gSKI()
for x in ca_details
if x.latest_ca_cert is not None)
for ski in skis_parent & skis_me:
logger.debug("Parent %s agrees that %s has SKI %s in resource class %s",
- parent.parent_handle, parent.self.self_handle, ski, rc.class_name)
+ parent.parent_handle, parent.self.self_handle, ski, class_name)
for ski in skis_parent - skis_me:
logger.debug("Parent %s thinks %s has SKI %s in resource class %s but I don't think so",
- parent.parent_handle, parent.self.self_handle, ski, rc.class_name)
+ parent.parent_handle, parent.self.self_handle, ski, class_name)
for ski in skis_me - skis_parent:
logger.debug("I think %s has SKI %s in resource class %s but parent %s doesn't think so",
- parent.self.self_handle, ski, rc.class_name, parent.parent_handle)
+ parent.self.self_handle, ski, class_name, parent.parent_handle)
if ca_details:
rpki.async.iterator(ca_details, loop, done)
else:
logger.warning("Existing resource class %s to %s from %s with no certificates, rekeying",
- rc.class_name, parent.self.self_handle, parent.parent_handle)
+ class_name, parent.self.self_handle, parent.parent_handle)
self.gctx.checkpoint()
self.rekey(cb, eb)
@@ -625,7 +633,7 @@ class ca_obj(rpki.sql.sql_persistent):
self = cls()
self.gctx = parent.gctx
self.parent_id = parent.parent_id
- self.parent_resource_class = rc.class_name
+ self.parent_resource_class = rc.get("class_name")
self.sql_store()
try:
self.sia_uri = self.construct_sia_uri(parent, rc)
@@ -634,18 +642,18 @@ class ca_obj(rpki.sql.sql_persistent):
raise
ca_detail = ca_detail_obj.create(self)
- def done(issue_response):
- c = issue_response.payload.classes[0].certs[0]
- logger.debug("CA %r received certificate %s", self, c.cert_url)
+ def done(r_msg):
+ c = r_msg[0][0]
+ logger.debug("CA %r received certificate %s", self, c.get("cert_url"))
ca_detail.activate(
ca = self,
- cert = c.cert,
- uri = c.cert_url,
+ cert = rpki.x509.X509(Base64 = c.text),
+ uri = c.get("cert_url"),
callback = cb,
errback = eb)
logger.debug("Sending issue request to %r from %r", parent, self.create)
- rpki.up_down.issue_pdu.query(parent, self, ca_detail, done, eb)
+ parent.up_down_issue_query(self, ca_detail, done, eb)
def delete(self, parent, callback):
"""
@@ -677,6 +685,7 @@ class ca_obj(rpki.sql.sql_persistent):
"""
Allocate a certificate serial number.
"""
+
self.last_issued_sn += 1
self.sql_mark_dirty()
return self.last_issued_sn
@@ -685,6 +694,7 @@ class ca_obj(rpki.sql.sql_persistent):
"""
Allocate a manifest serial number.
"""
+
self.last_manifest_sn += 1
self.sql_mark_dirty()
return self.last_manifest_sn
@@ -693,6 +703,7 @@ class ca_obj(rpki.sql.sql_persistent):
"""
Allocate a CRL serial number.
"""
+
self.last_crl_sn += 1
self.sql_mark_dirty()
return self.last_crl_sn
@@ -709,19 +720,19 @@ class ca_obj(rpki.sql.sql_persistent):
old_detail = self.active_ca_detail
new_detail = ca_detail_obj.create(self)
- def done(issue_response):
- c = issue_response.payload.classes[0].certs[0]
- logger.debug("CA %r received certificate %s", self, c.cert_url)
+ def done(r_msg):
+ c = r_msg[0][0]
+ logger.debug("CA %r received certificate %s", self, c.get("cert_url"))
new_detail.activate(
ca = self,
- cert = c.cert,
- uri = c.cert_url,
+ cert = rpki.x509.X509(Base64 = c.text),
+ uri = c.get("cert_url"),
predecessor = old_detail,
callback = cb,
errback = eb)
logger.debug("Sending issue request to %r from %r", parent, self.rekey)
- rpki.up_down.issue_pdu.query(parent, self, new_detail, done, eb)
+ parent.up_down_issue_query(self, new_detail, done, eb)
def revoke(self, cb, eb, revoke_all = False):
"""
@@ -783,6 +794,7 @@ class ca_detail_obj(rpki.sql.sql_persistent):
"""
Extra assertions for SQL decode of a ca_detail_obj.
"""
+
rpki.sql.sql_persistent.sql_decode(self, vals)
assert self.public_key is None or self.private_key_id is None or self.public_key.get_DER() == self.private_key_id.get_public_DER()
assert self.manifest_public_key is None or self.manifest_private_key_id is None or self.manifest_public_key.get_DER() == self.manifest_private_key_id.get_public_DER()
@@ -793,12 +805,14 @@ class ca_detail_obj(rpki.sql.sql_persistent):
"""
Fetch CA object to which this ca_detail links.
"""
+
return ca_obj.sql_fetch(self.gctx, self.ca_id)
def fetch_child_certs(self, child = None, ski = None, unique = False, unpublished = None):
"""
Fetch all child_cert objects that link to this ca_detail.
"""
+
return rpki.rpkid.child_cert_obj.fetch(self.gctx, child, self, ski, unique, unpublished)
@property
@@ -806,6 +820,7 @@ class ca_detail_obj(rpki.sql.sql_persistent):
"""
Fetch all child_cert objects that link to this ca_detail.
"""
+
return self.fetch_child_certs()
def unpublished_child_certs(self, when):
@@ -813,6 +828,7 @@ class ca_detail_obj(rpki.sql.sql_persistent):
Fetch all unpublished child_cert objects linked to this ca_detail
with attempted publication dates older than when.
"""
+
return self.fetch_child_certs(unpublished = when)
@property
@@ -820,6 +836,7 @@ class ca_detail_obj(rpki.sql.sql_persistent):
"""
Fetch all revoked_cert objects that link to this ca_detail.
"""
+
return revoked_cert_obj.sql_fetch_where(self.gctx, "ca_detail_id = %s", (self.ca_detail_id,))
@property
@@ -827,6 +844,7 @@ class ca_detail_obj(rpki.sql.sql_persistent):
"""
Fetch all ROA objects that link to this ca_detail.
"""
+
return rpki.rpkid.roa_obj.sql_fetch_where(self.gctx, "ca_detail_id = %s", (self.ca_detail_id,))
def unpublished_roas(self, when):
@@ -834,6 +852,7 @@ class ca_detail_obj(rpki.sql.sql_persistent):
Fetch all unpublished ROA objects linked to this ca_detail with
attempted publication dates older than when.
"""
+
return rpki.rpkid.roa_obj.sql_fetch_where(self.gctx, "ca_detail_id = %s AND published IS NOT NULL and published < %s", (self.ca_detail_id, when))
@property
@@ -841,27 +860,39 @@ class ca_detail_obj(rpki.sql.sql_persistent):
"""
Fetch all Ghostbuster objects that link to this ca_detail.
"""
+
return rpki.rpkid.ghostbuster_obj.sql_fetch_where(self.gctx, "ca_detail_id = %s", (self.ca_detail_id,))
+ def unpublished_ghostbusters(self, when):
+ """
+ Fetch all unpublished Ghostbusters objects linked to this
+ ca_detail with attempted publication dates older than when.
+ """
+
+ return rpki.rpkid.ghostbuster_obj.sql_fetch_where(self.gctx, "ca_detail_id = %s AND published IS NOT NULL and published < %s", (self.ca_detail_id, when))
+
@property
def ee_certificates(self):
"""
Fetch all EE certificate objects that link to this ca_detail.
"""
+
return rpki.rpkid.ee_cert_obj.sql_fetch_where(self.gctx, "ca_detail_id = %s", (self.ca_detail_id,))
- def unpublished_ghostbusters(self, when):
+ def unpublished_ee_certificates(self, when):
"""
- Fetch all unpublished Ghostbusters objects linked to this
+ Fetch all unpublished EE certificate objects linked to this
ca_detail with attempted publication dates older than when.
"""
- return rpki.rpkid.ghostbuster_obj.sql_fetch_where(self.gctx, "ca_detail_id = %s AND published IS NOT NULL and published < %s", (self.ca_detail_id, when))
+
+ return rpki.rpkid.ee_cert_obj.sql_fetch_where(self.gctx, "ca_detail_id = %s AND published IS NOT NULL and published < %s", (self.ca_detail_id, when))
@property
def crl_uri(self):
"""
Return publication URI for this ca_detail's CRL.
"""
+
return self.ca.sia_uri + self.crl_uri_tail
@property
@@ -869,6 +900,7 @@ class ca_detail_obj(rpki.sql.sql_persistent):
"""
Return tail (filename portion) of publication URI for this ca_detail's CRL.
"""
+
return self.public_key.gSKI() + ".crl"
@property
@@ -876,12 +908,14 @@ class ca_detail_obj(rpki.sql.sql_persistent):
"""
Return publication URI for this ca_detail's manifest.
"""
+
return self.ca.sia_uri + self.public_key.gSKI() + ".mft"
def has_expired(self):
"""
Return whether this ca_detail's certificate has expired.
"""
+
return self.latest_ca_cert.getNotAfter() <= rpki.sundial.now()
def covers(self, target):
@@ -901,7 +935,7 @@ class ca_detail_obj(rpki.sql.sql_persistent):
publisher = publication_queue()
self.latest_ca_cert = cert
- self.ca_cert_uri = uri.rsync()
+ self.ca_cert_uri = uri
self.generate_manifest_cert()
self.state = "active"
self.generate_crl(publisher = publisher)
@@ -933,11 +967,10 @@ class ca_detail_obj(rpki.sql.sql_persistent):
repository = ca.parent.repository
handler = False if allow_failure else None
for child_cert in self.child_certs:
- publisher.withdraw(cls = rpki.publication.certificate_elt,
- uri = child_cert.uri,
- obj = child_cert.cert,
- repository = repository,
- handler = handler)
+ publisher.queue(uri = child_cert.uri,
+ old_obj = child_cert.cert,
+ repository = repository,
+ handler = handler)
child_cert.sql_mark_deleted()
for roa in self.roas:
roa.revoke(publisher = publisher, allow_failure = allow_failure, fast = True)
@@ -948,21 +981,19 @@ class ca_detail_obj(rpki.sql.sql_persistent):
except AttributeError:
latest_manifest = None
if latest_manifest is not None:
- publisher.withdraw(cls = rpki.publication.manifest_elt,
- uri = self.manifest_uri,
- obj = self.latest_manifest,
- repository = repository,
- handler = handler)
+ publisher.queue(uri = self.manifest_uri,
+ old_obj = self.latest_manifest,
+ repository = repository,
+ handler = handler)
try:
latest_crl = self.latest_crl
except AttributeError:
latest_crl = None
if latest_crl is not None:
- publisher.withdraw(cls = rpki.publication.crl_elt,
- uri = self.crl_uri,
- obj = self.latest_crl,
- repository = repository,
- handler = handler)
+ publisher.queue(uri = self.crl_uri,
+ old_obj = self.latest_crl,
+ repository = repository,
+ handler = handler)
self.gctx.sql.sweep()
for cert in self.revoked_certs: # + self.child_certs
logger.debug("Deleting %r", cert)
@@ -995,13 +1026,18 @@ class ca_detail_obj(rpki.sql.sql_persistent):
ca = self.ca
parent = ca.parent
+ class_name = ca.parent_resource_class
+ gski = self.latest_ca_cert.gSKI()
def parent_revoked(r_msg):
- if r_msg.payload.ski != self.latest_ca_cert.gSKI():
+ if r_msg[0].get("class_name") != class_name:
+ raise rpki.exceptions.ResourceClassMismatch
+
+ if r_msg[0].get("ski") != gski:
raise rpki.exceptions.SKIMismatch
- logger.debug("Parent revoked %s, starting cleanup", self.latest_ca_cert.gSKI())
+ logger.debug("Parent revoked %s, starting cleanup", gski)
crl_interval = rpki.sundial.timedelta(seconds = parent.self.crl_interval)
@@ -1039,8 +1075,9 @@ class ca_detail_obj(rpki.sql.sql_persistent):
self.sql_mark_dirty()
publisher.call_pubd(cb, eb)
- logger.debug("Asking parent to revoke CA certificate %s", self.latest_ca_cert.gSKI())
- rpki.up_down.revoke_pdu.query(ca, self.latest_ca_cert.gSKI(), parent_revoked, eb)
+ logger.debug("Asking parent to revoke CA certificate %s", gski)
+ parent.up_down_revoke_query(class_name, gski, parent_revoked, eb)
+
def update(self, parent, ca, rc, sia_uri_changed, old_resources, callback, errback):
"""
@@ -1048,24 +1085,27 @@ class ca_detail_obj(rpki.sql.sql_persistent):
children of this ca_detail.
"""
- def issued(issue_response):
- c = issue_response.payload.classes[0].certs[0]
- logger.debug("CA %r received certificate %s", self, c.cert_url)
+ def issued(r_msg):
+ c = r_msg[0][0]
+ cert = rpki.x509.X509(Base64 = c.text)
+ cert_url = c.get("cert_url")
+
+ logger.debug("CA %r received certificate %s", self, cert_url)
if self.state == "pending":
return self.activate(
ca = ca,
- cert = c.cert,
- uri = c.cert_url,
+ cert = cert,
+ uri = cert_url,
callback = callback,
errback = errback)
- validity_changed = self.latest_ca_cert is None or self.latest_ca_cert.getNotAfter() != c.cert.getNotAfter()
+ validity_changed = self.latest_ca_cert is None or self.latest_ca_cert.getNotAfter() != cert.getNotAfter()
publisher = publication_queue()
- if self.latest_ca_cert != c.cert:
- self.latest_ca_cert = c.cert
+ if self.latest_ca_cert != cert:
+ self.latest_ca_cert = cert
self.sql_mark_dirty()
self.generate_manifest_cert()
self.generate_crl(publisher = publisher)
@@ -1093,7 +1133,8 @@ class ca_detail_obj(rpki.sql.sql_persistent):
publisher.call_pubd(callback, errback)
logger.debug("Sending issue request to %r from %r", parent, self.update)
- rpki.up_down.issue_pdu.query(parent, ca, self, issued, errback)
+ parent.up_down_issue_query(ca, self, issued, errback)
+
@classmethod
def create(cls, ca):
@@ -1172,6 +1213,7 @@ class ca_detail_obj(rpki.sql.sql_persistent):
notAfter = resources.valid_until)
if child_cert is None:
+ old_cert = None
child_cert = rpki.rpkid.child_cert_obj(
gctx = child.gctx,
child_id = child.child_id,
@@ -1179,6 +1221,7 @@ class ca_detail_obj(rpki.sql.sql_persistent):
cert = cert)
logger.debug("Created new child_cert %r", child_cert)
else:
+ old_cert = child_cert.cert
child_cert.cert = cert
del child_cert.ca_detail
child_cert.ca_detail_id = self.ca_detail_id
@@ -1187,10 +1230,10 @@ class ca_detail_obj(rpki.sql.sql_persistent):
child_cert.ski = cert.get_SKI()
child_cert.published = rpki.sundial.now()
child_cert.sql_store()
- publisher.publish(
- cls = rpki.publication.certificate_elt,
+ publisher.queue(
uri = child_cert.uri,
- obj = child_cert.cert,
+ old_obj = old_cert,
+ new_obj = child_cert.cert,
repository = ca.parent.repository,
handler = child_cert.published_callback)
self.generate_manifest(publisher = publisher)
@@ -1221,6 +1264,8 @@ class ca_detail_obj(rpki.sql.sql_persistent):
certlist.append((revoked_cert.serial, revoked_cert.revoked))
certlist.sort()
+ old_crl = self.latest_crl
+
self.latest_crl = rpki.x509.CRL.generate(
keypair = self.private_key_id,
issuer = self.latest_ca_cert,
@@ -1231,10 +1276,10 @@ class ca_detail_obj(rpki.sql.sql_persistent):
self.crl_published = rpki.sundial.now()
self.sql_mark_dirty()
- publisher.publish(
- cls = rpki.publication.crl_elt,
+ publisher.queue(
uri = self.crl_uri,
- obj = self.latest_crl,
+ old_obj = old_crl,
+ new_obj = self.latest_crl,
repository = parent.repository,
handler = self.crl_published_callback)
@@ -1242,7 +1287,8 @@ class ca_detail_obj(rpki.sql.sql_persistent):
"""
Check result of CRL publication.
"""
- pdu.raise_if_error()
+
+ rpki.publication.raise_if_error(pdu)
self.crl_published = None
self.sql_mark_dirty()
@@ -1278,6 +1324,7 @@ class ca_detail_obj(rpki.sql.sql_persistent):
objs.extend((e.uri_tail, e.cert) for e in self.ee_certificates)
logger.debug("Building manifest object %s", uri)
+ old_manifest = self.latest_manifest
self.latest_manifest = rpki.x509.SignedManifest.build(
serial = ca.next_manifest_number(),
thisUpdate = now,
@@ -1290,17 +1337,18 @@ class ca_detail_obj(rpki.sql.sql_persistent):
self.manifest_published = rpki.sundial.now()
self.sql_mark_dirty()
- publisher.publish(cls = rpki.publication.manifest_elt,
- uri = uri,
- obj = self.latest_manifest,
- repository = parent.repository,
- handler = self.manifest_published_callback)
+ publisher.queue(uri = uri,
+ old_obj = old_manifest,
+ new_obj = self.latest_manifest,
+ repository = parent.repository,
+ handler = self.manifest_published_callback)
def manifest_published_callback(self, pdu):
"""
Check result of manifest publication.
"""
- pdu.raise_if_error()
+
+ rpki.publication.raise_if_error(pdu)
self.manifest_published = None
self.sql_mark_dirty()
@@ -1361,21 +1409,19 @@ class ca_detail_obj(rpki.sql.sql_persistent):
self.crl_published is not None and \
self.crl_published < stale:
logger.debug("Retrying publication for %s", self.crl_uri)
- publisher.publish(cls = rpki.publication.crl_elt,
- uri = self.crl_uri,
- obj = self.latest_crl,
- repository = repository,
- handler = self.crl_published_callback)
+ publisher.queue(uri = self.crl_uri,
+ new_obj = self.latest_crl,
+ repository = repository,
+ handler = self.crl_published_callback)
if self.latest_manifest is not None and \
self.manifest_published is not None and \
self.manifest_published < stale:
logger.debug("Retrying publication for %s", self.manifest_uri)
- publisher.publish(cls = rpki.publication.manifest_elt,
- uri = self.manifest_uri,
- obj = self.latest_manifest,
- repository = repository,
- handler = self.manifest_published_callback)
+ publisher.queue(uri = self.manifest_uri,
+ new_obj = self.latest_manifest,
+ repository = repository,
+ handler = self.manifest_published_callback)
if not check_all:
return
@@ -1385,31 +1431,37 @@ class ca_detail_obj(rpki.sql.sql_persistent):
for child_cert in self.unpublished_child_certs(stale):
logger.debug("Retrying publication for %s", child_cert)
- publisher.publish(
- cls = rpki.publication.certificate_elt,
+ publisher.queue(
uri = child_cert.uri,
- obj = child_cert.cert,
+ new_obj = child_cert.cert,
repository = repository,
handler = child_cert.published_callback)
for roa in self.unpublished_roas(stale):
logger.debug("Retrying publication for %s", roa)
- publisher.publish(
- cls = rpki.publication.roa_elt,
+ publisher.queue(
uri = roa.uri,
- obj = roa.roa,
+ new_obj = roa.roa,
repository = repository,
handler = roa.published_callback)
for ghostbuster in self.unpublished_ghostbusters(stale):
logger.debug("Retrying publication for %s", ghostbuster)
- publisher.publish(
- cls = rpki.publication.ghostbuster_elt,
+ publisher.queue(
uri = ghostbuster.uri,
- obj = ghostbuster.ghostbuster,
+ new_obj = ghostbuster.ghostbuster,
repository = repository,
handler = ghostbuster.published_callback)
+ for ee_cert in self.unpublished_ee_certificates(stale):
+ logger.debug("Retrying publication for %s", ee_cert)
+ publisher.queue(
+ uri = ee_cert.uri,
+ new_obj = ee_cert.cert,
+ repository = repository,
+ handler = ee_cert.published_callback)
+
+
class child_cert_obj(rpki.sql.sql_persistent):
"""
Certificate that has been issued to a child.
@@ -1436,6 +1488,7 @@ class child_cert_obj(rpki.sql.sql_persistent):
"""
Initialize a child_cert_obj.
"""
+
rpki.sql.sql_persistent.__init__(self)
self.gctx = gctx
self.child_id = child_id
@@ -1451,6 +1504,7 @@ class child_cert_obj(rpki.sql.sql_persistent):
"""
Fetch child object to which this child_cert object links.
"""
+
return rpki.left_right.child_elt.sql_fetch(self.gctx, self.child_id)
@property
@@ -1459,6 +1513,7 @@ class child_cert_obj(rpki.sql.sql_persistent):
"""
Fetch ca_detail object to which this child_cert object links.
"""
+
return ca_detail_obj.sql_fetch(self.gctx, self.ca_detail_id)
@ca_detail.deleter
@@ -1473,6 +1528,7 @@ class child_cert_obj(rpki.sql.sql_persistent):
"""
Return the tail (filename) portion of the URI for this child_cert.
"""
+
return self.cert.gSKI() + ".cer"
@property
@@ -1480,6 +1536,7 @@ class child_cert_obj(rpki.sql.sql_persistent):
"""
Return the publication URI for this child_cert.
"""
+
return self.ca_detail.ca.sia_uri + self.uri_tail
def revoke(self, publisher, generate_crl_and_manifest = True):
@@ -1491,10 +1548,9 @@ class child_cert_obj(rpki.sql.sql_persistent):
ca = ca_detail.ca
logger.debug("Revoking %r %r", self, self.uri)
revoked_cert_obj.revoke(cert = self.cert, ca_detail = ca_detail)
- publisher.withdraw(
- cls = rpki.publication.certificate_elt,
- uri = self.uri,
- obj = self.cert,
+ publisher.queue(
+ uri = self.uri,
+ old_obj = self.cert,
repository = ca.parent.repository)
self.gctx.sql.sweep()
self.sql_delete()
@@ -1625,7 +1681,8 @@ class child_cert_obj(rpki.sql.sql_persistent):
"""
Publication callback: check result and mark published.
"""
- pdu.raise_if_error()
+
+ rpki.publication.raise_if_error(pdu)
self.published = None
self.sql_mark_dirty()
@@ -1649,6 +1706,7 @@ class revoked_cert_obj(rpki.sql.sql_persistent):
"""
Initialize a revoked_cert_obj.
"""
+
rpki.sql.sql_persistent.__init__(self)
self.gctx = gctx
self.serial = serial
@@ -1664,6 +1722,7 @@ class revoked_cert_obj(rpki.sql.sql_persistent):
"""
Fetch ca_detail object to which this revoked_cert_obj links.
"""
+
return ca_detail_obj.sql_fetch(self.gctx, self.ca_detail_id)
@classmethod
@@ -1671,6 +1730,7 @@ class revoked_cert_obj(rpki.sql.sql_persistent):
"""
Revoke a certificate.
"""
+
return cls(
serial = cert.getSerial(),
expires = cert.getNotAfter(),
@@ -1712,6 +1772,7 @@ class roa_obj(rpki.sql.sql_persistent):
"""
Fetch ca_detail object to which this roa_obj links.
"""
+
return rpki.rpkid.ca_detail_obj.sql_fetch(self.gctx, self.ca_detail_id)
@ca_detail.deleter
@@ -1725,6 +1786,7 @@ class roa_obj(rpki.sql.sql_persistent):
"""
Extra SQL fetch actions for roa_obj -- handle prefix lists.
"""
+
for version, datatype, attribute in ((4, rpki.resource_set.roa_prefix_set_ipv4, "ipv4"),
(6, rpki.resource_set.roa_prefix_set_ipv6, "ipv6")):
setattr(self, attribute, datatype.from_sql(
@@ -1739,6 +1801,7 @@ class roa_obj(rpki.sql.sql_persistent):
"""
Extra SQL insert actions for roa_obj -- handle prefix lists.
"""
+
for version, prefix_set in ((4, self.ipv4), (6, self.ipv6)):
if prefix_set:
self.gctx.sql.executemany(
@@ -1753,6 +1816,7 @@ class roa_obj(rpki.sql.sql_persistent):
"""
Extra SQL delete actions for roa_obj -- handle prefix lists.
"""
+
self.gctx.sql.execute("DELETE FROM roa_prefix WHERE roa_id = %s", (self.roa_id,))
def __repr__(self):
@@ -1894,10 +1958,9 @@ class roa_obj(rpki.sql.sql_persistent):
self.sql_store()
logger.debug("Generating %r URI %s", self, self.uri)
- publisher.publish(
- cls = rpki.publication.roa_elt,
+ publisher.queue(
uri = self.uri,
- obj = self.roa,
+ new_obj = self.roa,
repository = ca.parent.repository,
handler = self.published_callback)
if not fast:
@@ -1908,7 +1971,8 @@ class roa_obj(rpki.sql.sql_persistent):
"""
Check publication result.
"""
- pdu.raise_if_error()
+
+ rpki.publication.raise_if_error(pdu)
self.published = None
self.sql_mark_dirty()
@@ -1942,9 +2006,10 @@ class roa_obj(rpki.sql.sql_persistent):
logger.debug("Withdrawing %r %s and revoking its EE cert", self, uri)
rpki.rpkid.revoked_cert_obj.revoke(cert = cert, ca_detail = ca_detail)
- publisher.withdraw(cls = rpki.publication.roa_elt, uri = uri, obj = roa,
- repository = ca_detail.ca.parent.repository,
- handler = False if allow_failure else None)
+ publisher.queue(uri = uri,
+ old_obj = roa,
+ repository = ca_detail.ca.parent.repository,
+ handler = False if allow_failure else None)
if not regenerate:
self.sql_mark_deleted()
@@ -1958,6 +2023,7 @@ class roa_obj(rpki.sql.sql_persistent):
"""
Reissue ROA associated with this roa_obj.
"""
+
if self.ca_detail is None:
self.generate(publisher = publisher, fast = fast)
else:
@@ -1967,6 +2033,7 @@ class roa_obj(rpki.sql.sql_persistent):
"""
Return publication URI for a public key.
"""
+
return self.ca_detail.ca.sia_uri + key.gSKI() + ".roa"
@property
@@ -1974,6 +2041,7 @@ class roa_obj(rpki.sql.sql_persistent):
"""
Return the publication URI for this roa_obj's ROA.
"""
+
return self.ca_detail.ca.sia_uri + self.uri_tail
@property
@@ -1982,6 +2050,7 @@ class roa_obj(rpki.sql.sql_persistent):
Return the tail (filename portion) of the publication URI for this
roa_obj's ROA.
"""
+
return self.cert.gSKI() + ".roa"
@@ -2024,6 +2093,7 @@ class ghostbuster_obj(rpki.sql.sql_persistent):
"""
Fetch self object to which this ghostbuster_obj links.
"""
+
return rpki.left_right.self_elt.sql_fetch(self.gctx, self.self_id)
@property
@@ -2032,6 +2102,7 @@ class ghostbuster_obj(rpki.sql.sql_persistent):
"""
Fetch ca_detail object to which this ghostbuster_obj links.
"""
+
return rpki.rpkid.ca_detail_obj.sql_fetch(self.gctx, self.ca_detail_id)
def __init__(self, gctx = None, self_id = None, ca_detail_id = None, vcard = None):
@@ -2097,10 +2168,9 @@ class ghostbuster_obj(rpki.sql.sql_persistent):
self.sql_store()
logger.debug("Generating Ghostbuster record %r", self.uri)
- publisher.publish(
- cls = rpki.publication.ghostbuster_elt,
+ publisher.queue(
uri = self.uri,
- obj = self.ghostbuster,
+ new_obj = self.ghostbuster,
repository = ca.parent.repository,
handler = self.published_callback)
if not fast:
@@ -2110,7 +2180,8 @@ class ghostbuster_obj(rpki.sql.sql_persistent):
"""
Check publication result.
"""
- pdu.raise_if_error()
+
+ rpki.publication.raise_if_error(pdu)
self.published = None
self.sql_mark_dirty()
@@ -2144,9 +2215,10 @@ class ghostbuster_obj(rpki.sql.sql_persistent):
logger.debug("Withdrawing %r %s and revoking its EE cert", self, uri)
rpki.rpkid.revoked_cert_obj.revoke(cert = cert, ca_detail = ca_detail)
- publisher.withdraw(cls = rpki.publication.ghostbuster_elt, uri = uri, obj = ghostbuster,
- repository = ca_detail.ca.parent.repository,
- handler = False if allow_failure else None)
+ publisher.queue(uri = uri,
+ old_obj = ghostbuster,
+ repository = ca_detail.ca.parent.repository,
+ handler = False if allow_failure else None)
if not regenerate:
self.sql_mark_deleted()
@@ -2160,6 +2232,7 @@ class ghostbuster_obj(rpki.sql.sql_persistent):
"""
Reissue Ghostbuster associated with this ghostbuster_obj.
"""
+
if self.ghostbuster is None:
self.generate(publisher = publisher, fast = fast)
else:
@@ -2169,6 +2242,7 @@ class ghostbuster_obj(rpki.sql.sql_persistent):
"""
Return publication URI for a public key.
"""
+
return self.ca_detail.ca.sia_uri + key.gSKI() + ".gbr"
@property
@@ -2176,6 +2250,7 @@ class ghostbuster_obj(rpki.sql.sql_persistent):
"""
Return the publication URI for this ghostbuster_obj's ghostbuster.
"""
+
return self.ca_detail.ca.sia_uri + self.uri_tail
@property
@@ -2184,6 +2259,7 @@ class ghostbuster_obj(rpki.sql.sql_persistent):
Return the tail (filename portion) of the publication URI for this
ghostbuster_obj's ghostbuster.
"""
+
return self.cert.gSKI() + ".gbr"
@@ -2221,6 +2297,7 @@ class ee_cert_obj(rpki.sql.sql_persistent):
"""
Fetch self object to which this ee_cert_obj links.
"""
+
return rpki.left_right.self_elt.sql_fetch(self.gctx, self.self_id)
@property
@@ -2229,6 +2306,7 @@ class ee_cert_obj(rpki.sql.sql_persistent):
"""
Fetch ca_detail object to which this ee_cert_obj links.
"""
+
return rpki.rpkid.ca_detail_obj.sql_fetch(self.gctx, self.ca_detail_id)
@ca_detail.deleter
@@ -2246,6 +2324,7 @@ class ee_cert_obj(rpki.sql.sql_persistent):
Although, really, one has to ask why we don't just store g(SKI)
in rpkid.sql instead of ski....
"""
+
return base64.urlsafe_b64encode(self.ski).rstrip("=")
@gski.setter
@@ -2257,6 +2336,7 @@ class ee_cert_obj(rpki.sql.sql_persistent):
"""
Return the publication URI for this ee_cert_obj.
"""
+
return self.ca_detail.ca.sia_uri + self.uri_tail
@property
@@ -2265,6 +2345,7 @@ class ee_cert_obj(rpki.sql.sql_persistent):
Return the tail (filename portion) of the publication URI for this
ee_cert_obj.
"""
+
return self.cert.gSKI() + ".cer"
@classmethod
@@ -2292,12 +2373,11 @@ class ee_cert_obj(rpki.sql.sql_persistent):
ca_detail_id = ca_detail.ca_detail_id,
cert = cert)
- publisher.publish(
- cls = rpki.publication.certificate_elt,
- uri = self.uri,
- obj = self.cert,
+ publisher.queue(
+ uri = self.uri,
+ new_obj = self.cert,
repository = ca.parent.repository,
- handler = self.published_callback)
+ handler = self.published_callback)
self.sql_store()
@@ -2316,10 +2396,9 @@ class ee_cert_obj(rpki.sql.sql_persistent):
ca = ca_detail.ca
logger.debug("Revoking %r %r", self, self.uri)
revoked_cert_obj.revoke(cert = self.cert, ca_detail = ca_detail)
- publisher.withdraw(cls = rpki.publication.certificate_elt,
- uri = self.uri,
- obj = self.cert,
- repository = ca.parent.repository)
+ publisher.queue(uri = self.uri,
+ old_obj = self.cert,
+ repository = ca.parent.repository)
self.gctx.sql.sweep()
self.sql_delete()
if generate_crl_and_manifest:
@@ -2401,12 +2480,12 @@ class ee_cert_obj(rpki.sql.sql_persistent):
self.sql_mark_dirty()
- publisher.publish(
- cls = rpki.publication.certificate_elt,
- uri = self.uri,
- obj = self.cert,
+ publisher.queue(
+ uri = self.uri,
+ old_obj = old_cert,
+ new_obj = self.cert,
repository = ca_detail.ca.parent.repository,
- handler = self.published_callback)
+ handler = self.published_callback)
if must_revoke:
revoked_cert_obj.revoke(cert = old_cert.cert, ca_detail = old_ca_detail)
@@ -2423,7 +2502,8 @@ class ee_cert_obj(rpki.sql.sql_persistent):
"""
Publication callback: check result and mark published.
"""
- pdu.raise_if_error()
+
+ rpki.publication.raise_if_error(pdu)
self.published = None
self.sql_mark_dirty()
@@ -2451,29 +2531,49 @@ class publication_queue(object):
if self.replace:
self.uris = {}
- def _add(self, uri, obj, repository, handler, make_pdu):
+ def queue(self, uri, repository, handler = None, old_obj = None, new_obj = None):
+
+ assert old_obj is not None or new_obj is not None
+ assert old_obj is None or isinstance(old_obj, rpki.x509.uri_dispatch(uri))
+ assert new_obj is None or isinstance(new_obj, rpki.x509.uri_dispatch(uri))
+
+ logger.debug("Queuing publication action: uri %s, old %r, new %r", uri, old_obj, new_obj)
+
+ # id(repository) may need to change to repository.peer_contact_uri
+ # once we convert from our custom SQL cache to Django ORM.
+
rid = id(repository)
if rid not in self.repositories:
self.repositories[rid] = repository
- self.msgs[rid] = rpki.publication.msg.query()
+ self.msgs[rid] = Element(rpki.publication.tag_msg, nsmap = rpki.publication.nsmap,
+ type = "query", version = rpki.publication.version)
+
if self.replace and uri in self.uris:
- logger.debug("Removing publication duplicate <%s %r %r>",
- self.uris[uri].action, self.uris[uri].uri, self.uris[uri].payload)
- self.msgs[rid].remove(self.uris.pop(uri))
- pdu = make_pdu(uri = uri, obj = obj)
+ logger.debug("Removing publication duplicate %r", self.uris[uri])
+ old_pdu = self.uris.pop(uri)
+ self.msgs[rid].remove(old_pdu)
+ hash = old_pdu.get("hash")
+ elif old_obj is None:
+ hash = None
+ else:
+ hash = rpki.x509.sha256(old_obj.get_DER()).encode("hex")
+
+ if new_obj is None:
+ pdu = SubElement(self.msgs[rid], rpki.publication.tag_withdraw, uri = uri, hash = hash)
+ else:
+ pdu = SubElement(self.msgs[rid], rpki.publication.tag_publish, uri = uri)
+ pdu.text = new_obj.get_Base64()
+ if hash is not None:
+ pdu.set("hash", hash)
+
if handler is not None:
- self.handlers[id(pdu)] = handler
- pdu.tag = id(pdu)
- self.msgs[rid].append(pdu)
+ tag = str(id(pdu))
+ self.handlers[tag] = handler
+ pdu.set("tag", tag)
+
if self.replace:
self.uris[uri] = pdu
- def publish(self, cls, uri, obj, repository, handler = None):
- return self._add( uri, obj, repository, handler, cls.make_publish)
-
- def withdraw(self, cls, uri, obj, repository, handler = None):
- return self._add( uri, obj, repository, handler, cls.make_withdraw)
-
def call_pubd(self, cb, eb):
def loop(iterator, rid):
logger.debug("Calling pubd[%r]", self.repositories[rid])
@@ -2488,5 +2588,5 @@ class publication_queue(object):
return sum(len(self.msgs[rid]) for rid in self.repositories)
def empty(self):
- assert (not self.msgs) == (self.size == 0)
+ assert (not self.msgs) == (self.size == 0), "Assertion failure: not self.msgs: %r, self.size %r" % (not self.msgs, self.size)
return not self.msgs
generated by cgit v1.2.3 (git 2.25.1) at 2025-05-12 00:42:51 +0000