diff options
Diffstat (limited to 'rp/utils')
| -rwxr-xr-x | rp/utils/find_roa | 233 | ||||
| -rwxr-xr-x | rp/utils/hashdir | 60 | ||||
| -rwxr-xr-x | rp/utils/print_roa | 89 | ||||
| -rwxr-xr-x | rp/utils/print_rpki_manifest | 44 | ||||
| -rwxr-xr-x | rp/utils/scan_roas | 67 | ||||
| -rwxr-xr-x | rp/utils/scan_routercerts | 39 | ||||
| -rwxr-xr-x | rp/utils/uri | 80 |
7 files changed, 311 insertions, 301 deletions
diff --git a/rp/utils/find_roa b/rp/utils/find_roa index 4cfcccac..9a387c6a 100755 --- a/rp/utils/find_roa +++ b/rp/utils/find_roa @@ -25,134 +25,137 @@ import os import argparse import rpki.POW import rpki.oids +import rpki.config def check_dir(s): - if os.path.isdir(s): - return os.path.abspath(s) - else: - raise argparse.ArgumentTypeError("%r is not a directory" % s) + if os.path.isdir(s): + return os.path.abspath(s) + else: + raise argparse.ArgumentTypeError("%r is not a directory" % s) def filename_to_uri(filename): - if not filename.startswith(args.rcynic_dir): - raise ValueError - return "rsync://" + filename[len(args.rcynic_dir):].lstrip("/") + if not filename.startswith(args.rcynic_dir): + raise ValueError + return "rsync://" + filename[len(args.rcynic_dir):].lstrip("/") def uri_to_filename(uri): - if not uri.startswith("rsync://"): - raise ValueError - return os.path.join(args.rcynic_dir, uri[len("rsync://"):]) + if not uri.startswith("rsync://"): + raise ValueError + return os.path.join(args.rcynic_dir, uri[len("rsync://"):]) class Prefix(object): - """ - One prefix parsed from the command line. - """ - - def __init__(self, val): - addr, length = val.split("/") - length, sep, maxlength = length.partition("-") # pylint: disable=W0612 - self.prefix = rpki.POW.IPAddress(addr) - self.length = int(length) - self.maxlength = int(maxlength) if maxlength else self.length - if self.maxlength < self.length or self.length < 0 or self.length > self.prefix.bits: - raise ValueError - if self.prefix & ((1 << (self.prefix.bits - self.length)) - 1) != 0: - raise ValueError - - def matches(self, roa): - return any(self.prefix == prefix and - self.length == length and - (not args.match_maxlength or - self.maxlength == maxlength or - (maxlength is None and - self.length == self.maxlength)) - for prefix, length, maxlength in roa.prefixes) - - -class ROA(rpki.POW.ROA): - """ - Aspects of a ROA that we care about. - """ - - @classmethod - def parse(cls, fn): - assert fn.startswith(args.rcynic_dir) - self = cls.derReadFile(fn) - self.fn = fn - self.extractWithoutVerifying() - v4, v6 = self.getPrefixes() - self.prefixes = (v4 or ()) + (v6 or ()) - return self - - @property - def uri(self): - return filename_to_uri(self.fn) - - @property - def formatted_prefixes(self): - for prefix in self.prefixes: - if prefix[2] is None or prefix[1] == prefix[2]: - yield "%s/%d" % (prefix[0], prefix[1]) - else: - yield "%s/%d-%d" % (prefix[0], prefix[1], prefix[2]) - - def __str__(self): - prefixes = " ".join(self.formatted_prefixes) - plural = "es" if " " in prefixes else "" - if args.show_inception: - return "signingTime %s ASN %s prefix%s %s" % (self.signingTime(), self.getASID(), plural, prefixes) - else: - return "ASN %s prefix%s %s" % (self.getASID(), plural, prefixes) - - def show(self): - print "%s %s" % (self, self.fn if args.show_filenames else self.uri) - - def show_expiration(self): - print self - x = self.certs()[0] - fn = self.fn - uri = self.uri - while uri is not None: - name = fn if args.show_filenames else uri - if args.show_inception: - print "notBefore", x.getNotBefore(), "notAfter", x.getNotAfter(), name - else: - print x.getNotAfter(), name - for uri in x.getAIA() or (): - if uri.startswith("rsync://"): - break - else: - break - fn = uri_to_filename(uri) - if not os.path.exists(fn): - print "***** MISSING ******", uri - break - x = rpki.POW.X509.derReadFile(fn) - print - - -parser = argparse.ArgumentParser(description = __doc__) -parser.add_argument("-a", "--all", action = "store_true", help = "show all ROAs, do no prefix matching at all") -parser.add_argument("-m", "--match-maxlength", action = "store_true", help = "pay attention to maxLength values") -parser.add_argument("-e", "--show-expiration", action = "store_true", help = "show ROA chain expiration dates") -parser.add_argument("-f", "--show-filenames", action = "store_true", help = "show filenames instead of URIs") -parser.add_argument("-i", "--show-inception", action = "store_true", help = "show inception dates") -parser.add_argument("rcynic_dir", type = check_dir, help = "rcynic authenticated output directory") -parser.add_argument("prefixes", type = Prefix, nargs = "*", help = "ROA prefix(es) to match") -args = parser.parse_args() + """ + One prefix parsed from the command line. + """ + + def __init__(self, val): + addr, length = val.split("/") + length, sep, maxlength = length.partition("-") # pylint: disable=W0612 + self.prefix = rpki.POW.IPAddress(addr) + self.length = int(length) + self.maxlength = int(maxlength) if maxlength else self.length + if self.maxlength < self.length or self.length < 0 or self.length > self.prefix.bits: + raise ValueError + if self.prefix & ((1 << (self.prefix.bits - self.length)) - 1) != 0: + raise ValueError + + def matches(self, roa): # pylint: disable=W0621 + return any(self.prefix == prefix and + self.length == length and + (not args.match_maxlength or + self.maxlength == maxlength or + (maxlength is None and + self.length == self.maxlength)) + for prefix, length, maxlength in roa.prefixes) + + +class ROA(rpki.POW.ROA): # pylint: disable=W0232 + """ + Aspects of a ROA that we care about. + """ + + @classmethod + def parse(cls, fn): # pylint: disable=W0621 + assert fn.startswith(args.rcynic_dir) + self = cls.derReadFile(fn) # pylint: disable=E1101 + self.fn = fn + self.extractWithoutVerifying() + v4, v6 = self.getPrefixes() + self.prefixes = (v4 or ()) + (v6 or ()) + return self + + @property + def uri(self): + return filename_to_uri(self.fn) # pylint: disable=E1101 + + @property + def formatted_prefixes(self): + for prefix in self.prefixes: # pylint: disable=E1101 + if prefix[2] is None or prefix[1] == prefix[2]: + yield "%s/%d" % (prefix[0], prefix[1]) + else: + yield "%s/%d-%d" % (prefix[0], prefix[1], prefix[2]) + + def __str__(self): + # pylint: disable=E1101 + prefixes = " ".join(self.formatted_prefixes) + plural = "es" if " " in prefixes else "" + if args.show_inception: + return "signingTime %s ASN %s prefix%s %s" % (self.signingTime(), self.getASID(), plural, prefixes) + else: + return "ASN %s prefix%s %s" % (self.getASID(), plural, prefixes) + + def show(self): + # pylint: disable=E1101 + print "%s %s" % (self, self.fn if args.show_filenames else self.uri) + + def show_expiration(self): + print self + x = self.certs()[0] # pylint: disable=E1101 + fn = self.fn # pylint: disable=E1101,W0621 + uri = self.uri + while uri is not None: + name = fn if args.show_filenames else uri + if args.show_inception: + print "notBefore", x.getNotBefore(), "notAfter", x.getNotAfter(), name + else: + print x.getNotAfter(), name + for uri in x.getAIA() or (): + if uri.startswith("rsync://"): + break + else: + break + fn = uri_to_filename(uri) + if not os.path.exists(fn): + print "***** MISSING ******", uri + break + x = rpki.POW.X509.derReadFile(fn) + print + + +cfg = rpki.config.argparser(doc = __doc__) +cfg.argparser.add_argument("-a", "--all", action = "store_true", help = "show all ROAs, do no prefix matching at all") +cfg.argparser.add_argument("-m", "--match-maxlength", action = "store_true", help = "pay attention to maxLength values") +cfg.argparser.add_argument("-e", "--show-expiration", action = "store_true", help = "show ROA chain expiration dates") +cfg.argparser.add_argument("-f", "--show-filenames", action = "store_true", help = "show filenames instead of URIs") +cfg.argparser.add_argument("-i", "--show-inception", action = "store_true", help = "show inception dates") +cfg.argparser.add_argument("rcynic_dir", type = check_dir, help = "rcynic authenticated output directory") +cfg.argparser.add_argument("prefixes", type = Prefix, nargs = "*", help = "ROA prefix(es) to match") +args = cfg.argparser.parse_args() # If there's some way to automate this in the parser, I don't know what it is, so just catch it here. if args.all != (not args.prefixes): - parser.error("--all and prefix list are mutually exclusive") + parser.error("--all and prefix list are mutually exclusive") for root, dirs, files in os.walk(args.rcynic_dir): - for fn in files: - if fn.endswith(".roa"): - roa = ROA.parse(os.path.join(root, fn)) - if args.all or any(prefix.matches(roa) for prefix in args.prefixes): - if args.show_expiration: - roa.show_expiration() - else: - roa.show() + for fn in files: + if fn.endswith(".roa"): + roa = ROA.parse(os.path.join(root, fn)) + if args.all or any(prefix.matches(roa) for prefix in args.prefixes): + if args.show_expiration: + roa.show_expiration() + else: + roa.show() diff --git a/rp/utils/hashdir b/rp/utils/hashdir index d3fe393c..c2c100b8 100755 --- a/rp/utils/hashdir +++ b/rp/utils/hashdir @@ -26,42 +26,40 @@ distributed as part of the repository system. import os import sys -import argparse -import rpki.POW +import rpki.config + +from rpki.rcynicdb.iterator import authenticated_objects def check_dir(s): - if os.path.isdir(s): - return os.path.abspath(s) - else: - raise argparse.ArgumentTypeError("%r is not a directory" % s) + if os.path.isdir(s): + return os.path.abspath(s) + else: + raise argparse.ArgumentTypeError("{!r} is not a directory".format(s)) -parser = argparse.ArgumentParser(description = __doc__) -parser.add_argument("-v", "--verbose", action = "store_true", help = "whistle while you work") -parser.add_argument("rcynic_dir", type = check_dir, help = "rcynic authenticated output directory") -parser.add_argument("output_dir", help = "name of output directory to create") -args = parser.parse_args() +cfg = rpki.config.argparser(doc = __doc__) +cfg.argparser.add_argument("-v", "--verbose", action = "store_true", help = "whistle while you work") +cfg.argparser.add_argument("rcynic_dir", nargs = "?", type = check_dir, help = "rcynic authenticated output directory") +cfg.argparser.add_argument("output_dir", help = "name of output directory to create") +args = cfg.argparser.parse_args() if not os.path.isdir(args.output_dir): - os.makedirs(args.output_dir) + os.makedirs(args.output_dir) -for root, dirs, files in os.walk(args.rcynic_dir): - for ifn in files: - ifn = os.path.join(root, ifn) - if ifn.endswith(".cer"): - obj = rpki.POW.X509.derReadFile(ifn) - fmt = "%08x.%%d" % obj.getSubjectHash() - elif ifn.endswith(".crl"): - obj = rpki.POW.CRL.derReadFile(ifn) - fmt = "%08x.r%%d" % obj.getIssuerHash() - else: - continue +def store(uri, obj, fmt): for i in xrange(1000000): - ofn = os.path.join(args.output_dir, fmt % i) - if not os.path.exists(ofn): - with open(ofn, "w") as f: - f.write(obj.pemWrite()) - if args.verbose: - print ofn, "<=", ifn - break + fn = os.path.join(args.output_dir, fmt.format(i)) + if os.path.exists(fn): + continue + with open(fn, "w") as f: + f.write(obj.pemWrite()) + if args.verbose: + print fn, "<=", uri + return else: - sys.exit("No path name available for %s (%s)" % (ifn, ofn)) + sys.exit("No path name available for {} ({})".format(uri, fn)) + +for uri, cer in authenticated_objects(uri_suffix = ".cer"): + store(uri, cer, "{:08x}.{{:d}}".format(cer.getSubjectHash())) + +for uri, crl in authenticated_objects(uri_suffix = ".crl"): + store(uri, crl, "{:08x}.r{{:d}}".format(crl.getIssuerHash())) diff --git a/rp/utils/print_roa b/rp/utils/print_roa index d5db0c3c..c5b7793a 100755 --- a/rp/utils/print_roa +++ b/rp/utils/print_roa @@ -21,53 +21,56 @@ Pretty-print the content of a ROA. Does NOT attempt to verify the signature. """ -import argparse +import rpki.config import rpki.POW -class ROA(rpki.POW.ROA): +class ROA(rpki.POW.ROA): # pylint: disable=W0232 - @staticmethod - def _format_prefix(prefix): - if prefix[2] is None or prefix[1] == prefix[2]: - return "%s/%d" % (prefix[0], prefix[1]) - else: - return "%s/%d-%d" % (prefix[0], prefix[1], prefix[2]) + v4_prefixes = None + v6_prefixes = None + + @staticmethod + def _format_prefix(p): + if p[2] in (None, p[1]): + return "%s/%d" % (p[0], p[1]) + else: + return "%s/%d-%d" % (p[0], p[1], p[2]) - def parse(self): - self.extractWithoutVerifying() - v4, v6 = self.getPrefixes() - self.v4_prefixes = [self._format_prefix(p) for p in (v4 or ())] - self.v6_prefixes = [self._format_prefix(p) for p in (v6 or ())] + def parse(self): + self.extractWithoutVerifying() # pylint: disable=E1101 + v4, v6 = self.getPrefixes() # pylint: disable=E1101 + self.v4_prefixes = [self._format_prefix(p) for p in (v4 or ())] + self.v6_prefixes = [self._format_prefix(p) for p in (v6 or ())] -parser = argparse.ArgumentParser(description = __doc__) -parser.add_argument("-b", "--brief", action = "store_true", help = "show only ASN and prefix(es)") -parser.add_argument("-c", "--cms", action = "store_true", help = "print text representation of entire CMS blob") -parser.add_argument("-s", "--signing-time", action = "store_true", help = "show SigningTime in brief mode") -parser.add_argument("roas", nargs = "+", type = ROA.derReadFile, help = "ROA(s) to print") -args = parser.parse_args() +cfg = rpki.config.argparser(doc = __doc__) +cfg.argparser.add_argument("--brief", action = "store_true", help = "show only ASN and prefix(es)") +cfg.argparser.add_argument("--cms", action = "store_true", help = "print text representation of entire CMS blob") +cfg.argparser.add_argument("--signing-time", action = "store_true", help = "show SigningTime in brief mode") +cfg.argparser.add_argument("roas", nargs = "+", type = ROA.derReadFile, help = "ROA(s) to print") # pylint: disable=E1101 +args = cfg.argparser.parse_args() for roa in args.roas: - roa.parse() - if args.brief: - if args.signing_time: - print roa.signingTime(), - print roa.getASID(), " ".join(roa.v4_prefixes + roa.v6_prefixes) - else: - print "ROA Version: ", roa.getVersion() - print "SigningTime: ", roa.signingTime() - print "asID: ", roa.getASID() - if roa.v4_prefixes: - print " addressFamily:", 1 - for p in roa.v4_prefixes: - print " IPAddress:", p - if roa.v6_prefixes: - print " addressFamily:", 2 - for p in roa.v6_prefixes: - print " IPAddress:", p - if args.cms: - print roa.pprint() - for cer in roa.certs(): - print cer.pprint() - for crl in roa.crls(): - print crl.pprint() - print + roa.parse() + if args.brief: + if args.signing_time: + print roa.signingTime(), + print roa.getASID(), " ".join(roa.v4_prefixes + roa.v6_prefixes) + else: + print "ROA Version: ", roa.getVersion() + print "SigningTime: ", roa.signingTime() + print "asID: ", roa.getASID() + if roa.v4_prefixes: + print " addressFamily:", 1 + for prefix in roa.v4_prefixes: + print " IPAddress:", prefix + if roa.v6_prefixes: + print " addressFamily:", 2 + for prefix in roa.v6_prefixes: + print " IPAddress:", prefix + if args.cms: + print roa.pprint() + for cer in roa.certs(): + print cer.pprint() + for crl in roa.crls(): + print crl.pprint() + print diff --git a/rp/utils/print_rpki_manifest b/rp/utils/print_rpki_manifest index 5ebc6356..74a3fbd4 100755 --- a/rp/utils/print_rpki_manifest +++ b/rp/utils/print_rpki_manifest @@ -21,30 +21,30 @@ Pretty-print the content of a manifest. Does NOT attempt to verify the signature. """ -import argparse +import rpki.config import rpki.POW import rpki.oids -parser = argparse.ArgumentParser(description = __doc__) -parser.add_argument("-c", "--cms", action = "store_true", help = "print text representation of entire CMS blob") -parser.add_argument("manifests", nargs = "+", type = rpki.POW.Manifest.derReadFile, help = "manifest(s) to print") -args = parser.parse_args() +cfg = rpki.config.argparser(doc = __doc__) +cfg.argparser.add_argument("--cms", action = "store_true", help = "print text representation of entire CMS blob") +cfg.argparser.add_argument("manifests", nargs = "+", type = rpki.POW.Manifest.derReadFile, help = "manifest(s) to print") +args = cfg.argparser.parse_args() for mft in args.manifests: - mft.extractWithoutVerifying() - print "Manifest Version:", mft.getVersion() - print "SigningTime: ", mft.signingTime() - print "Number: ", mft.getManifestNumber() - print "thisUpdate: ", mft.getThisUpdate() - print "nextUpdate: ", mft.getNextUpdate() - print "fileHashAlg: ", rpki.oids.oid2name(mft.getAlgorithm()) - for i, fah in enumerate(mft.getFiles()): - name, hash = fah - print "fileList[%3d]: %s %s" % (i, ":".join(("%02X" % ord(h) for h in hash)), name) - if args.cms: - print mft.pprint() - for cer in mft.certs(): - print cer.pprint() - for crl in mft.crls(): - print crl.pprint() - print + mft.extractWithoutVerifying() + print "Manifest Version:", mft.getVersion() + print "SigningTime: ", mft.signingTime() + print "Number: ", mft.getManifestNumber() + print "thisUpdate: ", mft.getThisUpdate() + print "nextUpdate: ", mft.getNextUpdate() + print "fileHashAlg: ", rpki.oids.oid2name(mft.getAlgorithm()) + for i, fah in enumerate(mft.getFiles()): + name, obj_hash = fah + print "fileList[%3d]: %s %s" % (i, ":".join(("%02X" % ord(h) for h in obj_hash)), name) + if args.cms: + print mft.pprint() + for cer in mft.certs(): + print cer.pprint() + for crl in mft.crls(): + print crl.pprint() + print diff --git a/rp/utils/scan_roas b/rp/utils/scan_roas index a1b64f01..510fd7a0 100755 --- a/rp/utils/scan_roas +++ b/rp/utils/scan_roas @@ -24,40 +24,39 @@ per line. import os import argparse + +import rpki.config import rpki.POW +from rpki.rcynicdb.iterator import authenticated_objects + def check_dir(d): - if not os.path.isdir(d): - raise argparse.ArgumentTypeError("%r is not a directory" % d) - return d - -class ROA(rpki.POW.ROA): - - @classmethod - def parse(cls, fn): - self = cls.derReadFile(fn) - self.extractWithoutVerifying() - return self - - @property - def prefixes(self): - v4, v6 = self.getPrefixes() - for prefix, length, maxlength in (v4 or ()) + (v6 or ()): - if maxlength is None or length == maxlength: - yield "%s/%d" % (prefix, length) - else: - yield "%s/%d-%d" % (prefix, length, maxlength) - - def __str__(self): - return "%s %s %s" % (self.signingTime(), self.getASID(), " ".join(self.prefixes)) - -parser = argparse.ArgumentParser(description = __doc__) -parser.add_argument("rcynic_dir", nargs = "+", type = check_dir, - help = "rcynic authenticated output directory") -args = parser.parse_args() - -for rcynic_dir in args.rcynic_dir: - for root, dirs, files in os.walk(rcynic_dir): - for fn in files: - if fn.endswith(".roa"): - print ROA.parse(os.path.join(root, fn)) + if not os.path.isdir(d): + raise argparse.ArgumentTypeError("%r is not a directory" % d) + return d + +class ROA(rpki.POW.ROA): # pylint: disable=W0232 + + @property + def prefixes(self): + v4, v6 = self.getPrefixes() # pylint: disable=E1101 + for prefix, length, maxlength in (v4 or ()) + (v6 or ()): + if maxlength is None or length == maxlength: + yield "%s/%d" % (prefix, length) + else: + yield "%s/%d-%d" % (prefix, length, maxlength) + + def __str__(self): + # pylint: disable=E1101 + return "%s %s %s" % (self.signingTime(), self.getASID(), " ".join(self.prefixes)) + +cfg = rpki.config.argparser(doc = __doc__) +cfg.argparser.add_argument("rcynic_dir", nargs = "?", type = check_dir, + help = "rcynic authenticated output directory") +args = cfg.argparser.parse_args() + +for uri, roa in authenticated_objects(args.rcynic_dir, + uri_suffix = ".roa", + class_map = dict(roa = ROA)): + roa.extractWithoutVerifying() + print roa diff --git a/rp/utils/scan_routercerts b/rp/utils/scan_routercerts index 081a6293..540a8e25 100755 --- a/rp/utils/scan_routercerts +++ b/rp/utils/scan_routercerts @@ -26,32 +26,29 @@ import base64 import argparse import rpki.POW import rpki.oids +import rpki.config -def check_dir(s): - if not os.path.isdir(s): - raise argparse.ArgumentTypeError("%r is not a directory" % s) - return s - -parser = argparse.ArgumentParser(description = __doc__) -parser.add_argument("rcynic_dir", type = check_dir, help = "rcynic authenticated output directory") -args = parser.parse_args() +from rpki.rcynicdb.iterator import authenticated_objects -for root, dirs, files in os.walk(args.rcynic_dir): - - for fn in files: +def check_dir(s): + if not os.path.isdir(s): + raise argparse.ArgumentTypeError("{!r} is not a directory".format(s)) + return s - if not fn.endswith(".cer"): - continue +cfg = rpki.config.argparser(doc = __doc__) +cfg.argparser.add_argument("rcynic_dir", nargs = "?", type = check_dir, + help = "rcynic authenticated output directory") +args = cfg.argparser.parse_args() - x = rpki.POW.X509.derReadFile(os.path.join(root, fn)) +for uri, cer in authenticated_objects(args.rcynic_dir, uri_suffix = ".cer"): - if rpki.oids.id_kp_bgpsec_router not in (x.getEKU() or ()): - continue + if rpki.oids.id_kp_bgpsec_router not in (cer.getEKU() or ()): + continue - sys.stdout.write(base64.urlsafe_b64encode(x.getSKI()).rstrip("=")) + sys.stdout.write(base64.urlsafe_b64encode(cer.getSKI()).rstrip("=")) - for min_asn, max_asn in x.getRFC3779()[0]: - for asn in xrange(min_asn, max_asn + 1): - sys.stdout.write(" %s" % asn) + for min_asn, max_asn in cer.getRFC3779()[0]: + for asn in xrange(min_asn, max_asn + 1): + sys.stdout.write(" {}".format(asn)) - sys.stdout.write(" %s\n" % base64.b64encode(x.getPublicKey().derWritePublic())) + sys.stdout.write(" {}\n".format(base64.b64encode(cer.getPublicKey().derWritePublic()))) diff --git a/rp/utils/uri b/rp/utils/uri index e72d5e0d..d3d9eebb 100755 --- a/rp/utils/uri +++ b/rp/utils/uri @@ -24,47 +24,57 @@ Input files must be in DER format and may be either X.509v3 certificates or CMS objects which contain X.509v3 certificates in the CMS wrapper. """ -import argparse +import rpki.config import rpki.POW class Certificate(object): - @staticmethod - def first_rsync(uris): - if uris is not None: - for uri in uris: - if uri.startswith("rsync://"): - return uri - return None + @staticmethod + def first_whatever(uris, prefix): + if uris is not None: + for uri in uris: + if uri.startswith(prefix): + return uri + return None - def __init__(self, fn): - try: - x = rpki.POW.X509.derReadFile(fn) - except: # pylint: disable=W0702 - try: - cms = rpki.POW.CMS.derReadFile(fn) - cms.extractWithoutVerifying() - x = cms.certs()[0] - except: - raise ValueError - sia = x.getSIA() or (None, None, None) - self.fn = fn - self.uris = ( - ("AIA:caIssuers", self.first_rsync(x.getAIA())), - ("SIA:caRepository", self.first_rsync(sia[0])), - ("SIA:rpkiManifest", self.first_rsync(sia[1])), - ("SIA:signedObject", self.first_rsync(sia[2])), - ("CRLDP", self.first_rsync(x.getCRLDP()))) + def first_rsync(self, uris): + return self.first_whatever(uris, "rsync://") - def __str__(self): - words = [self.fn] if args.single_line else ["File: " + self.fn] - words.extend(" %s: %s" % (tag, uri) for tag, uri in self.uris if uri is not None) - return ("" if args.single_line else "\n").join(words) + def first_https(self, uris): + return self.first_whatever(uris, "https://") -parser = argparse.ArgumentParser(description = __doc__) -parser.add_argument("-s", "--single-line", action = "store_true", help = "single output line per object") -parser.add_argument("certs", nargs = "+", type = Certificate, help = "RPKI objects to examine") -args = parser.parse_args() + def first_http(self, uris): + return self.first_whatever(uris, "http://") + + def __init__(self, fn): + try: + x = rpki.POW.X509.derReadFile(fn) + except: + try: + cms = rpki.POW.CMS.derReadFile(fn) + cms.extractWithoutVerifying() + x = cms.certs()[0] + except: + raise ValueError + sia = x.getSIA() or (None, None, None, None) + self.fn = fn + self.uris = ( + ("AIA:caIssuers", self.first_rsync(x.getAIA())), + ("SIA:caRepository", self.first_rsync(sia[0])), + ("SIA:rpkiManifest", self.first_rsync(sia[1])), + ("SIA:signedObject", self.first_rsync(sia[2])), + ("SIA:rpkiNotify", self.first_https(sia[3]) or self.first_http(sia[3])), + ("CRLDP", self.first_rsync(x.getCRLDP()))) + + def __str__(self): + words = [self.fn] if args.single_line else ["File: " + self.fn] + words.extend(" %s: %s" % (tag, uri) for tag, uri in self.uris if uri is not None) + return ("" if args.single_line else "\n").join(words) + +cfg = rpki.config.argparser(doc = __doc__) +cfg.argparser.add_argument("-s", "--single-line", action = "store_true", help = "single output line per object") +cfg.argparser.add_argument("certs", nargs = "+", type = Certificate, help = "RPKI objects to examine") +args = cfg.argparser.parse_args() for cert in args.certs: - print cert + print cert |