starred/certbot-certbot
1
0
Fork 0
mirror of https://github.com/certbot/certbot.git synced 2026-07-28 00:35:50 +02:00
Code Issues Packages Projects Releases Wiki Activity

Various bug, docstring, and PEP8 fixes.

Browse Source
This commit is contained in:
Adam Woodbeck
2014-11-29 11:15:56 -05:00
parent c0c731b3e6
commit 5854d42672
3 changed files with 107 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files
letsencrypt/client/client.py
+76 -49
View File
@@ -1,3 +1,4 @@
"""ACME protocol client class and helper functions."""
import csv
import json
import os
@@ -33,6 +34,21 @@ class Client(object):
def __init__(self, ca_server, cert_signing_request=None,
private_key=None, use_curses=True):
"""
:param ca_server: Certificate authority server
:type ca_server: str
:param cert_signing_request: Contents of the CSR
:type cert_signing_request: str
:param private_key: Contents of the private key
:type private_key: str
:param use_curses: Use curses UI
:type use_curses: bool
"""
self.curses = use_curses
# Logger needs to be initialized before Configurator
@@ -42,25 +58,17 @@ class Client(object):
# line arg or client function to discover
self.config = apache_configurator.ApacheConfigurator(
CONFIG.SERVER_ROOT)
self.server = ca_server
if cert_signing_request:
self.csr_file = cert_signing_request
else:
self.csr_file = None
if private_key:
self.key_file = private_key
else:
self.key_file = None
self.csr = cert_signing_request
self.privkey = private_key
# TODO: Figure out all exceptions from this function
try:
self._validate_csr_key_cli()
except Exception as e:
except Exception as exc:
# TODO: Something nice here...
logger.fatal(("%s - until the programmers get their act together, "
"we are just going to exit" % str(e)))
"we are just going to exit" % str(exc)))
sys.exit(1)
self.server_url = "https://%s/acme/" % self.server
@@ -69,11 +77,16 @@ class Client(object):
:param domains: List of domains
:type domains: list
:param redirect:
:type redirect: bool|None
:param eula: EULA accepted
:type eula: bool
:raise Exception: CSR does not contain one of the specified names.
:raises errors.LetsEncryptClientError: CSR does not contain one of the
specified names.
"""
domains = [] if domains is None else domains
@@ -112,9 +125,9 @@ class Client(object):
_, csr_der = self.get_key_csr_pem()
# TODO: Handle this exception/problem
if not crypto_util.csr_matches_names(self.csr_file, self.names):
raise Exception(("CSR subject does not contain one of the "
"specified names"))
if not crypto_util.csr_matches_names(self.csr, self.names):
raise errors.LetsEncryptClientError(("CSR subject does not contain "
"one of the specified names"))
# Perform Challenges
responses, challenge_objs = self.verify_identity(challenge_msg)
@@ -165,7 +178,7 @@ class Client(object):
"""
auth_dict = self.send(acme.authorization_request(
challenge_msg["sessionID"], self.names[0],
challenge_msg["nonce"], responses, self.key_file))
challenge_msg["nonce"], responses, self.privkey))
try:
return self.is_expected_msg(auth_dict, "authorization")
@@ -188,7 +201,7 @@ class Client(object):
"""
logger.info("Preparing and sending CSR..")
return self.send_and_receive_expected(
acme.certificate_request(csr_der, self.key_file), "certificate")
acme.certificate_request(csr_der, self.privkey), "certificate")
def acme_revocation(self, cert):
"""Handle ACME "revocation" phase.
@@ -221,14 +234,14 @@ class Client(object):
:param msg: ACME message (JSON serializable).
:type msg: dict
:raises: TypeError if `msg` is not JSON serializable or
jsonschema.ValidationError if not valid ACME message or
`errors.LetsEncryptClientError` in case of connection error
or if response from server is not a valid ACME message.
:returns: Server response message.
:rtype: dict
:raises TypeError: if `msg` is not JSON serializable
:raises jsonschema.ValidationError: if not valid ACME message
:raises errors.LetsEncryptClientError: in case of connection error
or if response from server is not a valid ACME message.
"""
json_encoded = json.dumps(msg)
acme.acme_object_validate(json_encoded)
@@ -266,6 +279,8 @@ class Client(object):
:returns: ACME response message of expected type.
:rtype: dict
:raises errors.LetsEncryptClientError: An exception is thrown
"""
response = self.send(msg)
try:
@@ -291,11 +306,11 @@ class Client(object):
reponse message.
:type rounds: int
:raises: Exception
:returns: ACME response message from server.
:rtype: dict
:raises errors.LetsEncryptClientError:
"""
for _ in xrange(rounds):
if response["type"] == expected:
@@ -322,6 +337,7 @@ class Client(object):
(rounds * delay))
def list_certs_keys(self):
"""List trusted Let's Encrypt certificates."""
list_file = os.path.join(CONFIG.CERT_KEY_BACKUP, "LIST")
certs = []
@@ -415,7 +431,7 @@ class Client(object):
for host in vhost:
self.config.deploy_cert(host,
os.path.abspath(cert_file),
os.path.abspath(self.key_file),
os.path.abspath(self.privkey),
cert_chain_abspath)
# Enable any vhost that was issued to, but not enabled
if not host.enabled:
@@ -506,6 +522,9 @@ class Client(object):
:param encrypt: Should the certificate key be encrypted?
:type encrypt: bool
:returns: True if key file was stored successfully, False otherwise.
:rtype: bool
"""
list_file = os.path.join(CONFIG.CERT_KEY_BACKUP, "LIST")
le_util.make_or_verify_dir(CONFIG.CERT_KEY_BACKUP, 0700)
@@ -523,22 +542,24 @@ class Client(object):
for row in csvreader:
idx = int(row[0]) + 1
csvwriter = csv.writer(csvfile)
csvwriter.writerow([str(idx), cert_file, self.key_file])
csvwriter.writerow([str(idx), cert_file, self.privkey])
else:
with open(list_file, 'wb') as csvfile:
csvwriter = csv.writer(csvfile)
csvwriter.writerow(["0", cert_file, self.key_file])
csvwriter.writerow(["0", cert_file, self.privkey])
shutil.copy2(self.key_file,
shutil.copy2(self.privkey,
os.path.join(
CONFIG.CERT_KEY_BACKUP,
os.path.basename(self.key_file) + "_" + str(idx)))
os.path.basename(self.privkey) + "_" + str(idx)))
shutil.copy2(cert_file,
os.path.join(
CONFIG.CERT_KEY_BACKUP,
os.path.basename(cert_file) + "_" + str(idx)))
return True
def redirect_to_ssl(self, vhost):
for ssl_vh in vhost:
success, redirect_vhost = self.config.enable_redirect(ssl_vh)
@@ -607,7 +628,7 @@ class Client(object):
challenge_objs.append({
"type": "dvsni",
"listSNITuple": sni_todo,
"dvsni_key": os.path.abspath(self.key_file),
"dvsni_key": os.path.abspath(self.privkey),
})
challenge_obj_indices.append(sni_satisfies)
logger.debug(sni_todo)
@@ -632,31 +653,34 @@ class Client(object):
:rtype: tuple
"""
key_pem = None
csr_pem = None
if not self.key_file:
self.key_file = crypto_util.make_key(CONFIG.RSA_KEY_SIZE)
if not self.privkey:
key_pem = crypto_util.make_key(CONFIG.RSA_KEY_SIZE)
self.privkey = key_pem
# Save file
le_util.make_or_verify_dir(CONFIG.KEY_DIR, 0700)
key_f, key_filename = le_util.unique_file(
os.path.join(CONFIG.KEY_DIR, "key-letsencrypt.pem"), 0600)
key_f.write(self.key_file)
key_f.write(self.privkey)
key_f.close()
logger.info("Generating key: %s" % key_filename)
else:
key_pem = self.privkey
if not self.csr:
csr_pem, csr_der = crypto_util.make_csr(self.privkey, self.names)
self.csr = csr_pem
if not self.csr_file:
self.csr_file, csr_der = crypto_util.make_csr(self.key_file,
self.names)
# Save CSR
le_util.make_or_verify_dir(CONFIG.CERT_DIR, 0755)
csr_f, csr_filename = le_util.unique_file(
os.path.join(CONFIG.CERT_DIR, "csr-letsencrypt.pem"), 0644)
csr_f.write(self.csr_file)
csr_f.write(self.csr)
csr_f.close()
logger.info("Creating CSR: %s" % csr_filename)
else:
csr = M2Crypto.X509.load_request_string(self.csr_file)
csr_pem, csr_der = csr.as_pem(), csr.as_der()
csr_obj = M2Crypto.X509.load_request_string(self.csr)
csr_pem, csr_der = csr_obj.as_pem(), csr_obj.as_der()
if csr_return_format == 'der':
return key_pem, csr_der
@@ -675,18 +699,21 @@ class Client(object):
# and allow the user to take more appropriate actions
# If CSR is provided, it must be readable and valid.
if self.csr_file and not crypto_util.valid_csr(self.csr_file):
raise Exception("The provided CSR is not a valid CSR")
if self.csr and not crypto_util.valid_csr(self.csr):
raise errors.LetsEncryptClientError("The provided CSR is not a "
"valid CSR")
# If key is provided, it must be readable and valid.
if self.key_file and not crypto_util.valid_privkey(self.key_file):
raise Exception("The provided key is not a valid key")
if self.privkey and not crypto_util.valid_privkey(self.privkey):
raise errors.LetsEncryptClientError("The provided key is not a "
"valid key")
# If CSR and key are provided, the key must be the same key used
# in the CSR.
if self.csr_file and self.key_file:
if not crypto_util.csr_matches_pubkey(self.csr_file, self.key_file):
raise Exception("The key and CSR do not match")
if self.csr and self.privkey:
if not crypto_util.csr_matches_pubkey(self.csr, self.privkey):
raise errors.LetsEncryptClientError("The key and CSR do not "
"match")
def get_all_names(self):
"""Return all valid names in the configuration."""
letsencrypt/client/crypto_util.py
+24 -23
View File
@@ -1,3 +1,4 @@
"""Let's Encrypt client crypto utility functions"""
import binascii
import hashlib
import time
@@ -206,33 +207,33 @@ def get_cert_info(filename):
# A. Do more checks to verify that the CSR is trusted/valid
# B. Audit the parsing code for vulnerabilities
def valid_csr(csr_file):
def valid_csr(csr):
"""Validate CSR.
Check if `csr_filename` is a valid CSR for the given domains.
Check if `csr` is a valid CSR for the given domains.
:param csr_file: CSR file contents
:type csr_file: str
:param csr: CSR file contents
:type csr: str
:returns: Validity of CSR.
:rtype: bool
"""
try:
csr = M2Crypto.X509.load_request_string(csr_file)
return bool(csr.verify(csr.get_pubkey()))
csr_obj = M2Crypto.X509.load_request_string(csr)
return bool(csr_obj.verify(csr_obj.get_pubkey()))
except M2Crypto.X509.X509Error:
return False
def csr_matches_names(csr_file, domains):
def csr_matches_names(csr, domains):
"""Check if CSR contains the subject of one of the domains.
M2Crypto currently does not expose the OpenSSL interface to
also check the SAN extension. This is insufficient for full testing
:param csr_file: CSR file contents
:type csr_file: str
:param csr: CSR file contents
:type csr: str
:param domains: Domains the CSR should contain.
:type domains: list
@@ -242,41 +243,41 @@ def csr_matches_names(csr_file, domains):
"""
try:
csr = M2Crypto.X509.load_request_string(csr_file)
return csr.get_subject().CN in domains
csr_obj = M2Crypto.X509.load_request_string(csr)
return csr_obj.get_subject().CN in domains
except M2Crypto.X509.X509Error:
return False
def valid_privkey(privkey_file):
def valid_privkey(privkey):
"""Is valid RSA private key?
:param privkey_file: Private key file contents
:type privkey_file: str
:param privkey: Private key file contents
:type privkey: str
:returns: Validity of private key.
:rtype: bool
"""
try:
return bool(M2Crypto.RSA.load_key_string(privkey_file).check_key())
return bool(M2Crypto.RSA.load_key_string(privkey).check_key())
except M2Crypto.RSA.RSAError:
return False
def csr_matches_pubkey(csr_file, privkey_file):
def csr_matches_pubkey(csr, privkey):
"""Does private key correspond to the subject public key in the CSR?
:param csr_file: CSR file contents
:type csr_file: str
:param csr: CSR file contents
:type csr: str
:param privkey_file: Private key file contents
:type privkey_file: str
:param privkey: Private key file contents
:type privkey: str
:returns: Correspondence of private key to CSR subject public key.
:rtype: bool
"""
csr = M2Crypto.X509.load_request_string(csr_file)
privkey = M2Crypto.RSA.load_key_string(privkey_file)
return csr.get_pubkey().get_rsa().pub() == privkey.pub()
csr_obj = M2Crypto.X509.load_request_string(csr)
privkey_obj = M2Crypto.RSA.load_key_string(privkey)
return csr_obj.get_pubkey().get_rsa().pub() == privkey_obj.pub()
letsencrypt/scripts/main.py
+7 -15
View File
@@ -60,10 +60,7 @@ def main():
parser.add_argument("--test", dest="test", action="store_true",
help="Run in test mode.")
try:
args = parser.parse_args()
except IOError as exc:
parser.error(exc)
args = parser.parse_args()
# Enforce '--privkey' is set along with '--csr'.
if args.csr and not args.privkey:
@@ -101,19 +98,14 @@ def read_file(filename):
:returns: File contents
:rtype: str
:raise IOError: File does not exist or is not readable. file() will
potentially throw its own IOError exceptions in addition to the two
explicitely thrown.
:raises argparse.ArgumentTypeError: File does not exist or is not readable.
"""
if not os.path.exists(filename):
raise IOError("the file '{0}' is not found".format(filename))
if not os.access(filename, os.R_OK):
raise IOError("the file '{0}' is not readable".format(filename))
return file(filename, 'rU').read()
try:
return file(filename, 'rU').read()
except IOError as exc:
raise argparse.ArgumentTypeError(exc.strerror)
def rollback(config, checkpoints):