First baby steps towards testing new rpki.POW extended validation code.

svn path=/branches/tk705/; revision=6178

2 files changed, 195 insertions, 7 deletions

diff --git a/ext/POW.c b/ext/POW.c
index 9bbb9eba..16dd070d 100644
--- a/ext/POW.c
+++ b/ext/POW.c
@@ -3923,8 +3923,6 @@ static char x509_object_verify__doc__[] =
 
 #warning Write real x509_object_verify__doc__[] once API is stable.
 
-/* Let's try signalling that it's a TA by having (len(trusted) == 1 and trusted[0] is self) */
-
 static PyObject *
 x509_object_verify(x509_object *self, PyObject *args, PyObject *kwds)
 {
@@ -3938,6 +3936,7 @@ x509_object_verify(x509_object *self, PyObject *args, PyObject *kwds)
   PyObject *untrusted = Py_None;
   PyObject *crl       = Py_None;
   PyObject *status    = Py_None;
+  X509 *issuer = NULL;
   int ok = 0, is_ta;
 
   if (!PyArg_ParseTupleAndKeywords(args, kwds, "|OOOO!O", kwlist,
@@ -3972,16 +3971,51 @@ x509_object_verify(x509_object *self, PyObject *args, PyObject *kwds)
   if (crl != Py_None)
     X509_VERIFY_PARAM_set_flags(ctx->ctx->param, X509_V_FLAG_CRL_CHECK);
 
+  /*
+   * Tedious search for issuer.  Should we even be doing this?  rcynic
+   * knows which cert it thinks is the issuer, so it's a waste of time
+   * there, and we don't need to do this when we're not doing detailed
+   * RPKI checking, so the answer is probably no, we don't need this.
+   *
+   * Except that it seems to work better.  Which may just mean that I
+   * hnorked the ordering of the trusted chain when passing it in
+   * during testing.
+   *
+   * For the moment, keep options open, clean up later.
+   */
+
+#if 1
+
+  {
+    int i;
+    for (i = 0; issuer == NULL && i < sk_X509_num(trusted_stack); i++)
+      if (X509_check_issued((issuer = sk_X509_value(trusted_stack, i)), self->x509) != 0)
+        issuer = NULL;
+    for (i = 0; issuer == NULL && i < sk_X509_num(untrusted_stack); i++)
+      if (X509_check_issued((issuer = sk_X509_value(untrusted_stack, i)), self->x509) != 0)
+        issuer = NULL;
+  }
+
+  is_ta = (sk_X509_num(trusted_stack) == 1 && 
+           sk_X509_num(untrusted_stack) == 0 &&
+           X509_cmp(issuer, self->x509) == 0);
+
+#else
+#warning Do we need to do something about picking issuer out of trusted_stack?
+
   is_ta = (sk_X509_num(trusted_stack) == 1 &&
            sk_X509_num(untrusted_stack) == 0 &&
            X509_cmp(sk_X509_value(trusted_stack, 0), self->x509) == 0 &&
-           X509_check_issued(self->x509, self->x509));
+           X509_check_issued(self->x509, self->x509) == 0);
 
-#warning Need to do something about picking issuer out of trusted_stack
-#warning Need to do something about check_object_type_*
+#endif
+
+  if (issuer == NULL)
+    issuer = sk_X509_value(trusted_stack, 0);
 
-  if (status != Py_None && !check_x509(self->x509, sk_X509_value(trusted_stack, 0), status,
-                                       is_ta, check_object_type_cer, ctx->ctx))
+#warning Need to do something about check_object_type_* mess
+
+  if (status != Py_None && !check_x509(self->x509, issuer, status, is_ta, check_object_type_cer, ctx->ctx))
     goto error;
 
 #warning Not sure about reference counts in this version.  Might be safest to stuff PyObject* for all inputs into ctx so that everything mentioned here is protected until user releases ctx.
@@ -7902,6 +7936,8 @@ cms_object_verify_helper(cms_object *self, PyObject *args, PyObject *kwds, PyObj
   if ((flags & ~flag_mask) != 0)
     lose_value_error("Bad CMS_verify() flags");
 
+  flags |= CMS_NO_SIGNER_CERT_VERIFY;
+
   if ((bio = BIO_new(BIO_s_mem())) == NULL)
     lose_no_memory();
 
diff --git a/rp/rcynic/rcynicng b/rp/rcynic/rcynicng
new file mode 100755
index 00000000..b9d23380
--- /dev/null
+++ b/rp/rcynic/rcynicng
@@ -0,0 +1,152 @@
+#!/usr/bin/env python
+
+# $Id$
+
+"""
+Reimplementation of rcynic in Python.  Work in progress.
+
+Well, OK, at the moment this doesn't even come close to being a
+replacement for the C version of rcynic, must less adding the new
+features that were the reason for bothering with all this.  Right now,
+this is just a test framework for the new POW.c code to support Python
+RP code.  Gotta start somewhere.
+"""
+
+import os
+import sys
+import time
+import argparse
+
+import rpki.POW
+
+from lxml.etree import ElementTree, Element, SubElement, Comment
+
+args = None
+
+def check_dir(s):
+    if not os.path.isdir(s):
+        raise argparse.ArgumentTypeError("%r is not a directory" % s)
+    return s
+
+def parse_options():
+    global args                         # pylint: disable=W0603
+    parser = argparse.ArgumentParser(description = __doc__)
+    parser.add_argument("--unauthenticated",   type = check_dir, default = "rcynic-data/unauthenticated")
+    parser.add_argument("--old-authenticated", type = check_dir, default = "rcynic-data/authenticated.old")
+    parser.add_argument("--tals",              type = check_dir, default = "sample-trust-anchors")
+    parser.add_argument("--output",                              default = "rcynic-data/rcynicng-output")
+    args = parser.parse_args()
+
+
+def read_tals():
+    for root, dirs, files in os.walk(args.tals):
+        for fn in files:
+            if fn.endswith(".tal"):
+                with open(os.path.join(root, fn), "r") as f:
+                    lines = f.readlines()
+                uri = lines.pop(0).strip()
+                b64 = "".join(lines[lines.index("\n"):])
+                key = rpki.POW.Asymmetric.derReadPublic(b64.decode("base64"))
+                yield uri, key
+
+def uri_to_fn(uri, base = None):
+    fn = uri[uri.index("://")+3:]
+    if base is not None:
+        fn = os.path.join(base, fn)
+    return fn
+
+def first_uri(uris, scheme):
+    for uri in uris:
+        if uri.startswith(scheme):
+            return uri
+    return None
+
+def first_rsync_uri(uris):
+    return first_uri(uris, "rsync://")
+
+
+def walk_tree(ca, trusted, crl, basedir):
+    ca_status = set()
+    ca.verify(trusted = trusted, crl = crl, status = ca_status)
+    trusted.insert(0, ca)
+    diruri, mfturi = [first_rsync_uri(uri) for uri in  ca.getSIA()[:2]]
+    mft = rpki.POW.Manifest.derReadFile(uri_to_fn(mfturi, basedir))
+    ee = mft.certs()[0]
+    crldp = first_rsync_uri(ee.getCRLDP())
+    crl = rpki.POW.CRL.derReadFile(uri_to_fn(crldp, basedir))
+    crl_status = set()
+    mft_status = set()
+    crl.verify(ca, crl_status)
+    ee.verify(trusted = trusted, crl = crl, status = mft_status)
+    mft.verify(status = mft_status)
+
+    print "CA status: ", ", ".join(str(s) for s in ca_status)
+    print "CRL status:", ", ".join(str(s) for s in crl_status)
+    print "MFT status:", ", ".join(str(s) for s in mft_status)
+
+    for fn, digest in mft.getFiles():
+
+        with open(os.path.join(uri_to_fn(diruri, basedir), fn), "rb") as f:
+            obj = f.read()
+        dgst = rpki.POW.Digest(rpki.POW.SHA256_DIGEST)
+        dgst.update(obj)
+        print fn, digest.encode("hex"), "OK hash" if dgst.digest() == digest else "Bad hash"
+
+        if fn.endswith(".crl") and obj != crl.derWrite():
+            print "CRL mismatch"
+        if fn.endswith(".crl"):
+            continue
+
+        if fn.endswith(".roa"):
+            roa = rpki.POW.ROA.derRead(obj)
+            roa_status = set()
+            ee = roa.certs()[0]
+            ee.verify(trusted = trusted, crl = crl, status = roa_status)
+            roa.verify(status = roa_status)
+            continue
+
+        if fn.endswith(".gbr"):
+            gbr = rpki.POW.CMS.derRead(obj)
+            gbr_status = set()
+            ee = gbr.certs()[0]
+            ee.verify(trusted = trusted, crl = crl, status = gbr_status)
+            vcard = gbr.verify(status = gbr_status)
+            print vcard
+            continue
+
+        cer = rpki.POW.X509.derRead(obj)
+        bc = cer.getBasicConstraints()
+        if bc and bc[0]:
+            try:
+                walk_tree(cer, trusted, crl, basedir)
+            except rpki.POW.Error as e:
+                print "CA", diruri + fn, "failed:", e
+        else:
+            cer_status = set()
+            cer.verify(trusted = trusted, crl = crl, status = cer_status)
+
+
+def main():
+
+    os.putenv("TZ", "UTC")
+    time.tzset()
+
+    parse_options()
+
+    basedir = args.unauthenticated
+
+    for uri, pk in read_tals():
+        print
+        try:
+            x = rpki.POW.X509.derReadFile(uri_to_fn(uri, basedir))
+        except rpki.POW.OpenSSLError:
+            print "Couldn't open TA {}".format(uri)
+        else:
+            ok = pk.derWritePublic() == x.getPublicKey().derWritePublic()
+            print "OK " if ok else "Bad", uri
+            if ok:
+                walk_tree(x, [x], None, basedir)
+
+
+if __name__ == "__main__":
+    main()