MCPcopy Create free account
hub / github.com/zappa/Zappa / _send_signed_request

Function _send_signed_request

zappa/letsencrypt.py:394–429  ·  view source on GitHub ↗

Helper function to make signed requests to Boulder

(url, payload)

Source from the content-addressed store, hash-verified

392
393
394def _send_signed_request(url, payload):
395 """
396 Helper function to make signed requests to Boulder
397 """
398 payload64 = _b64(json.dumps(payload).encode("utf8"))
399
400 out = parse_account_key()
401 header = get_boulder_header(out)
402
403 protected = copy.deepcopy(header)
404 protected["nonce"] = urlopen(DEFAULT_CA + "/directory").headers["Replay-Nonce"]
405 protected64 = _b64(json.dumps(protected).encode("utf8"))
406 cmd = [
407 "openssl",
408 "dgst",
409 "-sha256",
410 "-sign",
411 os.path.join(gettempdir(), "account.key"),
412 ]
413 proc = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
414 out, err = proc.communicate("{0}.{1}".format(protected64, payload64).encode("utf8"))
415 if proc.returncode != 0: # pragma: no cover
416 raise IOError("OpenSSL Error: {0}".format(err))
417 data = json.dumps(
418 {
419 "header": header,
420 "protected": protected64,
421 "payload": payload64,
422 "signature": _b64(out),
423 }
424 )
425 try:
426 resp = urlopen(url, data.encode("utf8"))
427 return resp.getcode(), resp.read()
428 except IOError as e:
429 return getattr(e, "code", None), getattr(e, "read", e.__str__)()
430
431
432##

Callers 3

register_accountFunction · 0.85
get_certFunction · 0.85
sign_certificateFunction · 0.85

Calls 5

_b64Function · 0.85
parse_account_keyFunction · 0.85
get_boulder_headerFunction · 0.85
gettempdirFunction · 0.85
communicateMethod · 0.80

Tested by

no test coverage detected