aboutsummaryrefslogtreecommitdiff
path: root/ca/tests/testpoke.py
diff options
context:
space:
mode:
authorRob Austein <sra@hactrn.net>2015-10-26 06:29:00 +0000
committerRob Austein <sra@hactrn.net>2015-10-26 06:29:00 +0000
commitb46deb1417dc3596e9ac9fe2fe8cc0b7f42457e7 (patch)
treeca0dc0276d1adc168bc3337ce0564c4ec4957c1b /ca/tests/testpoke.py
parent397beaf6d9900dc3b3cb612c89ebf1d57b1d16f6 (diff)
"Any programmer who fails to comply with the standard naming, formatting,
or commenting conventions should be shot. If it so happens that it is inconvenient to shoot him, then he is to be politely requested to recode his program in adherence to the above standard." -- Michael Spier, Digital Equipment Corporation svn path=/branches/tk705/; revision=6152
Diffstat (limited to 'ca/tests/testpoke.py')
-rw-r--r--ca/tests/testpoke.py118
1 files changed, 59 insertions, 59 deletions
diff --git a/ca/tests/testpoke.py b/ca/tests/testpoke.py
index 68f967b9..60cc5690 100644
--- a/ca/tests/testpoke.py
+++ b/ca/tests/testpoke.py
@@ -58,80 +58,80 @@ yaml_data = yaml.load(args.yaml)
yaml_cmd = args.request
if yaml_cmd is None and len(yaml_data["requests"]) == 1:
- yaml_cmd = yaml_data["requests"].keys()[0]
+ yaml_cmd = yaml_data["requests"].keys()[0]
yaml_req = yaml_data["requests"][yaml_cmd]
def get_PEM(name, cls, y = yaml_data):
- if name in y:
- return cls(PEM = y[name])
- if name + "-file" in y:
- return cls(PEM_file = y[name + "-file"])
- return None
+ if name in y:
+ return cls(PEM = y[name])
+ if name + "-file" in y:
+ return cls(PEM_file = y[name + "-file"])
+ return None
def get_PEM_chain(name, cert = None):
- chain = []
- if cert is not None:
- chain.append(cert)
- if name in yaml_data:
- chain.extend(rpki.x509.X509(PEM = x) for x in yaml_data[name])
- elif name + "-file" in yaml_data:
- chain.extend(rpki.x509.X509(PEM_file = x) for x in yaml_data[name + "-file"])
- return chain
+ chain = []
+ if cert is not None:
+ chain.append(cert)
+ if name in yaml_data:
+ chain.extend(rpki.x509.X509(PEM = x) for x in yaml_data[name])
+ elif name + "-file" in yaml_data:
+ chain.extend(rpki.x509.X509(PEM_file = x) for x in yaml_data[name + "-file"])
+ return chain
def query_up_down(q_pdu):
- q_msg = rpki.up_down.message_pdu.make_query(
- payload = q_pdu,
- sender = yaml_data["sender-id"],
- recipient = yaml_data["recipient-id"])
- q_der = rpki.up_down.cms_msg_saxify().wrap(q_msg, cms_key, cms_certs, cms_crl)
-
- def done(r_der):
- global last_cms_timestamp
- r_cms = rpki.up_down.cms_msg_saxify(DER = r_der)
- r_msg = r_cms.unwrap([cms_ta] + cms_ca_certs)
- last_cms_timestamp = r_cms.check_replay(last_cms_timestamp)
- print r_cms.pretty_print_content()
- try:
- r_msg.payload.check_response()
- except (rpki.async.ExitNow, SystemExit):
- raise
- except Exception, e:
- fail(e)
-
- rpki.http.want_persistent_client = False
-
- rpki.http.client(
- msg = q_der,
- url = yaml_data["posturl"],
- callback = done,
- errback = fail,
- content_type = rpki.up_down.content_type)
+ q_msg = rpki.up_down.message_pdu.make_query(
+ payload = q_pdu,
+ sender = yaml_data["sender-id"],
+ recipient = yaml_data["recipient-id"])
+ q_der = rpki.up_down.cms_msg_saxify().wrap(q_msg, cms_key, cms_certs, cms_crl)
+
+ def done(r_der):
+ global last_cms_timestamp
+ r_cms = rpki.up_down.cms_msg_saxify(DER = r_der)
+ r_msg = r_cms.unwrap([cms_ta] + cms_ca_certs)
+ last_cms_timestamp = r_cms.check_replay(last_cms_timestamp)
+ print r_cms.pretty_print_content()
+ try:
+ r_msg.payload.check_response()
+ except (rpki.async.ExitNow, SystemExit):
+ raise
+ except Exception, e:
+ fail(e)
+
+ rpki.http.want_persistent_client = False
+
+ rpki.http.client(
+ msg = q_der,
+ url = yaml_data["posturl"],
+ callback = done,
+ errback = fail,
+ content_type = rpki.up_down.content_type)
def do_list():
- query_up_down(rpki.up_down.list_pdu())
+ query_up_down(rpki.up_down.list_pdu())
def do_issue():
- q_pdu = rpki.up_down.issue_pdu()
- req_key = get_PEM("cert-request-key", rpki.x509.RSA, yaml_req) or cms_key
- q_pdu.class_name = yaml_req["class"]
- q_pdu.pkcs10 = rpki.x509.PKCS10.create(
- keypair = req_key,
- is_ca = True,
- caRepository = yaml_req["sia"][0],
- rpkiManifest = yaml_req["sia"][0] + req_key.gSKI() + ".mft")
- query_up_down(q_pdu)
+ q_pdu = rpki.up_down.issue_pdu()
+ req_key = get_PEM("cert-request-key", rpki.x509.RSA, yaml_req) or cms_key
+ q_pdu.class_name = yaml_req["class"]
+ q_pdu.pkcs10 = rpki.x509.PKCS10.create(
+ keypair = req_key,
+ is_ca = True,
+ caRepository = yaml_req["sia"][0],
+ rpkiManifest = yaml_req["sia"][0] + req_key.gSKI() + ".mft")
+ query_up_down(q_pdu)
def do_revoke():
- q_pdu = rpki.up_down.revoke_pdu()
- q_pdu.class_name = yaml_req["class"]
- q_pdu.ski = yaml_req["ski"]
- query_up_down(q_pdu)
+ q_pdu = rpki.up_down.revoke_pdu()
+ q_pdu.class_name = yaml_req["class"]
+ q_pdu.ski = yaml_req["ski"]
+ query_up_down(q_pdu)
dispatch = { "list" : do_list, "issue" : do_issue, "revoke" : do_revoke }
def fail(e): # pylint: disable=W0621
- sys.exit("Testpoke failed: %s" % e)
+ sys.exit("Testpoke failed: %s" % e)
cms_ta = get_PEM("cms-ca-cert", rpki.x509.X509)
cms_cert = get_PEM("cms-cert", rpki.x509.X509)
@@ -143,7 +143,7 @@ cms_ca_certs = get_PEM_chain("cms-ca-certs")
last_cms_timestamp = None
try:
- dispatch[yaml_req["type"]]()
- rpki.async.event_loop()
+ dispatch[yaml_req["type"]]()
+ rpki.async.event_loop()
except Exception, e:
- fail(e)
+ fail(e)