aboutsummaryrefslogtreecommitdiff
path: root/rpkid/rpki/rcynic.py
diff options
context:
space:
mode:
Diffstat (limited to 'rpkid/rpki/rcynic.py')
-rw-r--r--rpkid/rpki/rcynic.py275
1 files changed, 0 insertions, 275 deletions
diff --git a/rpkid/rpki/rcynic.py b/rpkid/rpki/rcynic.py
deleted file mode 100644
index 73394fb8..00000000
--- a/rpkid/rpki/rcynic.py
+++ /dev/null
@@ -1,275 +0,0 @@
-# Copyright (C) 2010-2012 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.
-
-"""
-Iterator class to parse the output of an rcynic run.
-"""
-
-__version__ = '$Id$'
-
-import os
-import rpki.x509
-import rpki.exceptions
-import rpki.resource_set
-from xml.etree.ElementTree import ElementTree
-
-class UnknownObject(rpki.exceptions.RPKI_Exception):
- """
- Unrecognized object in rcynic result cache.
- """
-
-class NotRsyncURI(rpki.exceptions.RPKI_Exception):
- """
- URI is not an rsync URI.
- """
-
-class rcynic_object(object):
- """
- An object read from rcynic cache.
- """
-
- def __init__(self, filename, **kwargs):
- self.filename = filename
- for k, v in kwargs.iteritems():
- setattr(self, k, v)
- self.obj = self.obj_class(DER_file = filename)
-
- def __repr__(self):
- return "<%s %s %s at 0x%x>" % (self.__class__.__name__, self.uri, self.resources, id(self))
-
- def show_attrs(self, *attrs):
- """
- Print a bunch of object attributes, quietly ignoring any that
- might be missing.
- """
- for a in attrs:
- try:
- print "%s: %s" % (a.capitalize(), getattr(self, a))
- except AttributeError:
- pass
-
- def show(self):
- """
- Print common object attributes.
- """
- self.show_attrs("filename", "uri", "status", "timestamp")
-
-class rcynic_certificate(rcynic_object):
- """
- A certificate from rcynic cache.
- """
-
- obj_class = rpki.x509.X509
-
- def __init__(self, filename, **kwargs):
- rcynic_object.__init__(self, filename, **kwargs)
- self.notBefore = self.obj.getNotBefore()
- self.notAfter = self.obj.getNotAfter()
- self.aia_uri = self.obj.get_aia_uri()
- self.sia_directory_uri = self.obj.get_sia_directory_uri()
- self.manifest_uri = self.obj.get_sia_manifest_uri()
- self.resources = self.obj.get_3779resources()
- self.is_ca = self.obj.is_CA()
- self.serial = self.obj.getSerial()
- self.issuer = self.obj.getIssuer()
- self.subject = self.obj.getSubject()
- self.ski = self.obj.hSKI()
- self.aki = self.obj.hAKI()
-
- def show(self):
- """
- Print certificate attributes.
- """
- rcynic_object.show(self)
- self.show_attrs("notBefore", "notAfter", "aia_uri", "sia_directory_uri", "resources")
-
-class rcynic_roa(rcynic_object):
- """
- A ROA from rcynic cache.
- """
-
- obj_class = rpki.x509.ROA
-
- def __init__(self, filename, **kwargs):
- rcynic_object.__init__(self, filename, **kwargs)
- self.obj.extract()
- self.asID = self.obj.get_POW().getASID()
- self.prefix_sets = []
- v4, v6 = self.obj.get_POW().getPrefixes()
- if v4:
- self.prefix_sets.append(rpki.resource_set.roa_prefix_set_ipv4([
- rpki.resource_set.roa_prefix_ipv4(p[0], p[1], p[2]) for p in v4]))
- if v6:
- self.prefix_sets.append(rpki.resource_set.roa_prefix_set_ipv6([
- rpki.resource_set.roa_prefix_ipv6(p[0], p[1], p[2]) for p in v6]))
- self.ee = rpki.x509.X509(POW = self.obj.get_POW().certs()[0])
- self.notBefore = self.ee.getNotBefore()
- self.notAfter = self.ee.getNotAfter()
- self.aia_uri = self.ee.get_aia_uri()
- self.resources = self.ee.get_3779resources()
- self.issuer = self.ee.getIssuer()
- self.serial = self.ee.getSerial()
- self.subject = self.ee.getSubject()
- self.aki = self.ee.hAKI()
- self.ski = self.ee.hSKI()
-
- def show(self):
- """
- Print ROA attributes.
- """
- rcynic_object.show(self)
- self.show_attrs("notBefore", "notAfter", "aia_uri", "resources", "asID")
- if self.prefix_sets:
- print "Prefixes:", ",".join(str(i) for i in self.prefix_sets)
-
-class rcynic_ghostbuster(rcynic_object):
- """
- Ghostbuster record from the rcynic cache.
- """
-
- obj_class = rpki.x509.Ghostbuster
-
- def __init__(self, *args, **kwargs):
- rcynic_object.__init__(self, *args, **kwargs)
- self.obj.extract()
- self.vcard = self.obj.get_content()
- self.ee = rpki.x509.X509(POW = self.obj.get_POW().certs()[0])
- self.notBefore = self.ee.getNotBefore()
- self.notAfter = self.ee.getNotAfter()
- self.aia_uri = self.ee.get_aia_uri()
- self.issuer = self.ee.getIssuer()
- self.serial = self.ee.getSerial()
- self.subject = self.ee.getSubject()
- self.aki = self.ee.hAKI()
- self.ski = self.ee.hSKI()
-
- def show(self):
- rcynic_object.show(self)
- self.show_attrs("notBefore", "notAfter", "vcard")
-
-file_name_classes = {
- ".cer" : rcynic_certificate,
- ".gbr" : rcynic_ghostbuster,
- ".roa" : rcynic_roa }
-
-class rcynic_file_iterator(object):
- """
- Iterate over files in an rcynic output tree, yielding a Python
- representation of each object found.
- """
-
- def __init__(self, rcynic_root,
- authenticated_subdir = "authenticated"):
- self.rcynic_dir = os.path.join(rcynic_root, authenticated_subdir)
-
- def __iter__(self):
- for root, dirs, files in os.walk(self.rcynic_dir): # pylint: disable=W0612
- for filename in files:
- filename = os.path.join(root, filename)
- ext = os.path.splitext(filename)[1]
- if ext in file_name_classes:
- yield file_name_classes[ext](filename)
-
-class validation_status_element(object):
- def __init__(self, *args, **kwargs):
- self.attrs = []
- for k, v in kwargs.iteritems():
- setattr(self, k, v)
- # attribute names are saved so that the __repr__ method can
- # display the subset of attributes the user specified
- self.attrs.append(k)
- self._obj = None
-
- def get_obj(self):
- if not self._obj:
- self._obj = self.file_class(filename=self.filename, uri=self.uri)
- return self._obj
-
- def __repr__(self):
- v = [self.__class__.__name__, 'id=%s' % str(id(self))]
- v.extend(['%s=%s' % (x, getattr(self, x)) for x in self.attrs])
- return '<%s>' % (' '.join(v),)
-
- obj = property(get_obj)
-
-class rcynic_xml_iterator(object):
- """
- Iterate over validation_status entries in the XML output from an
- rcynic run. Yields a tuple for each entry:
-
- timestamp, generation, status, object
-
- where URI, status, and timestamp are the corresponding values from
- the XML element, OK is a boolean indicating whether validation was
- considered succesful, and object is a Python representation of the
- object in question. If OK is True, object will be from rcynic's
- authenticated output tree; otherwise, object will be from rcynic's
- unauthenticated output tree.
-
- Note that it is possible for the same URI to appear in more than one
- validation_status element; in such cases, the succesful case (OK
- True) should be the last entry (as rcynic will stop trying once it
- gets a good copy), but there may be multiple failures, which might
- or might not have different status codes.
- """
-
- def __init__(self, rcynic_root, xml_file,
- authenticated_old_subdir = "authenticated.old",
- unauthenticated_subdir = "unauthenticated"):
- self.rcynic_root = rcynic_root
- self.xml_file = xml_file
- self.authenticated_subdir = os.path.join(rcynic_root, 'authenticated')
- self.authenticated_old_subdir = os.path.join(rcynic_root, authenticated_old_subdir)
- self.unauthenticated_subdir = os.path.join(rcynic_root, unauthenticated_subdir)
-
- base_uri = "rsync://"
-
- def uri_to_filename(self, uri):
- if uri.startswith(self.base_uri):
- return uri[len(self.base_uri):]
- else:
- raise NotRsyncURI, "Not an rsync URI %r" % uri
-
- def __iter__(self):
- for validation_status in ElementTree(file=self.xml_file).getroot().getiterator("validation_status"):
- timestamp = validation_status.get("timestamp")
- status = validation_status.get("status")
- uri = validation_status.text.strip()
- generation = validation_status.get("generation")
-
- # determine the path to this object
- if status == 'object_accepted':
- d = self.authenticated_subdir
- elif generation == 'backup':
- d = self.authenticated_old_subdir
- else:
- d = self.unauthenticated_subdir
-
- filename = os.path.join(d, self.uri_to_filename(uri))
-
- ext = os.path.splitext(filename)[1]
- if ext in file_name_classes:
- yield validation_status_element(timestamp = timestamp, generation = generation,
- uri=uri, status = status, filename = filename,
- file_class = file_name_classes[ext])
-
-def label_iterator(xml_file):
- """
- Returns an iterator which contains all defined labels from an rcynic XML
- output file. Each item is a tuple of the form
- (label, kind, description).
- """
-
- for label in ElementTree(file=xml_file).find("labels"):
- yield label.tag, label.get("kind"), label.text.strip()
generated by cgit v1.2.3 (git 2.25.1) at 2025-05-29 23:00:41 +0000