factorise cli_flask code, add more comments to functions, add admin revoke crt by serial

    Split cli_flask functions used to renew, sign and revoke certificate.
    Allow to revoke a certificate by serial PUT /crt/revoke/serial, this
    method required admin authentication. Also add GET /crt/serial/<string:serial>

caucase/ca.py

@@ -158,6 +158,8 @@ class CertificateAuthority(object):
   def getPendingCertificateRequest(self, csr_id):
     """
     Retrieve the content of a pending signing request.
+    
+    @param csr_id: The id of CSR returned by the storage
     """
     return self._storage.getPendingCertificateRequest(csr_id)
 
@@ -165,6 +167,8 @@ class CertificateAuthority(object):
     """
     Sanity-check CSR, stores it and generates a unique signing request
     identifier (crt_id).
+    
+    @param csr: CSR string in PEM format
     """
     # Check number of already-pending signing requests
     # Check if csr is self-signed
@@ -191,12 +195,18 @@ class CertificateAuthority(object):
   def deletePendingCertificateRequest(self, csr_id):
     """
     Reject a pending certificate signing request.
+    
+    @param csr_id: The id of CSR returned by the storage
     """
     self._storage.deletePendingCertificateRequest(csr_id)
 
   def getPendingCertificateRequestList(self, limit=0, with_data=False):
     """
-      Return list of signed certificate
+      Return list of pending certificate signature request
+      
+      @param limit: number of element to fetch, 0 is not limit (int)
+      @param with_data: True or False, say if return csr PEM string associated
+        to others informations (bool).
     """
     return self._storage.getPendingCertificateRequestList(limit, with_data)
 
@@ -204,6 +214,11 @@ class CertificateAuthority(object):
     """
       Generate new signed certificate. `ca_key_pair` is the CA key_pair to use
         if None, use the latest CA key_pair
+      
+      @param csr_id: CSR ID returned by storage, csr should be linked to the 
+          new certificate (string).
+      @param ca_key_pair: The CA key_pair to used for signature. If None, the
+          latest key_pair is used.
     """
     # Apply extensions (ex: "not a certificate", ...)
     # Generate a certificate from the CSR
@@ -220,11 +235,33 @@ class CertificateAuthority(object):
     return crt_id
 
   def getCertificate(self, crt_id):
+    """
+      Return a Certificate string in PEM format
+      
+      @param crt_id: Certificate ID returned by storage during certificate creation
+    """
     return self._storage.getCertificate(crt_id)
 
+  def getCertificateFromSerial(self, serial):
+    """
+      Return a Certificate string in PEM format
+      
+      @param serial: serial of the certificate (string)
+    """
+    cert =  self._storage.getCertificateFromSerial(serial)
+    if not cert.content:
+      raise NotFound('Content certificate with serial %r is not found.' % (
+                      serial,
+                    ))
+    return cert.content
+
   def getSignedCertificateList(self, limit=0, with_data=False):
     """
       Return list of signed certificate
+      
+      @param limit: number of element to fetch, 0 is not limit (int)
+      @param with_data: True or False, say if return cert PEM string associated
+        to others informations (bool).
     """
     return self._storage.getSignedCertificateList(limit, with_data)
 
@@ -236,7 +273,7 @@ class CertificateAuthority(object):
 
   def getValidCACertificateChain(self):
     """
-    Return the ca certificate chain for all valid certificates
+      Return the ca certificate chain for all valid certificates with key
     """
     result = []
     iter_key_pair = iter(self._ca_key_pairs_list)
@@ -251,6 +288,8 @@ class CertificateAuthority(object):
   def getCAKeypairForCertificate(self, cert):
     """
       Return the nearest CA key_pair to the next extension date of the cert
+      
+      @param cert: X509 certificate
     """
     cert_valid_date = datetime.strptime(cert.get_notAfter(), '%Y%m%d%H%M%SZ')
     next_valid_date =  datetime.utcnow() + timedelta(0, self.crt_life_time)
@@ -274,6 +313,20 @@ class CertificateAuthority(object):
     return selected_keypair
 
   def revokeCertificate(self, wrapped_crt):
+    """
+      Revoke a certificate
+      
+      @param wrapped_crt: The revoke request dict containing certificate to
+        revoke and signature algorithm used to sign the request.
+        {
+          "signature": "signature string for payload",
+          "digest": "Signature algorithm (ex: SHA256"),
+          "payload": dict of data: {
+              "revoke_crt": "Certificate to revoke",
+              "reason": "Revoke reason"
+            }
+        }
+    """
     payload = utils.unwrap(wrapped_crt, lambda x: x['revoke_crt'], self.digest_list)
 
     try:
@@ -287,18 +340,36 @@ class CertificateAuthority(object):
                 "The CA couldn't reconize the certificate to revoke.")
 
     crt = self._loadCertificate(payload['revoke_crt'])
-    expiration_date = datetime.strptime(crt.get_notAfter(), '%Y%m%d%H%M%SZ')
-    expire_in = expiration_date  - datetime.now()
-    if crt.has_expired():
-      raise ExpiredCertificate("Could not revoke a certificate which has expired" \
-              "since %r days." % -1*expire_in.days)
     reason = payload['reason']
     return self._storage.revokeCertificate(
               utils.getSerialToInt(crt),
-              expiration_date,
               reason)
 
+  def revokeCertificateFromSerial(self, serial):
+    """
+      Directly revoke a certificate from serial
+      
+      @param serial: The serial of the certificate (int)
+    """
+    return self._storage.revokeCertificate(
+              serial,
+              reason="")
+
   def renew(self, wrapped_csr):
+    """
+      Renew a certificate
+      
+      @param wrapped_csr: The revoke request dict containing certificate to
+        revoke and signature algorithm used to sign the request.
+        {
+          "signature": "signature string for payload",
+          "digest": "Signature algorithm (ex: SHA256"),
+          "payload": dict of data: {
+              "crt": "Old certificate to renew",
+              "renew_csr": "New CSR to sign"
+            }
+        }
+    """
     payload = utils.unwrap(wrapped_csr, lambda x: x['crt'], self.digest_list)
     csr = payload['renew_csr']
 
@@ -377,8 +448,8 @@ class CertificateAuthority(object):
     # Here comes the actual certificate
     serial = self._storage.getNextCertificateSerialNumber()
     cert = crypto.X509()
-    # 3 = v3
-    cert.set_version(3)
+    # version v3
+    cert.set_version(2)
     cert.set_serial_number(serial)
     cert.gmtime_adj_notBefore(0)
     cert.gmtime_adj_notAfter(self.crt_life_time)

caucase/storage.py

@@ -24,7 +24,7 @@ from datetime import datetime, timedelta
 from OpenSSL import crypto
 from caucase import db
 from caucase import utils
-from caucase.exceptions import (NoStorage, NotFound, Found)
+from caucase.exceptions import (NoStorage, NotFound, Found, ExpiredCertificate)
 from flask_user import UserMixin
 
 STATUS_VALIDATED = 'validated'
@@ -326,7 +326,7 @@ class Storage(object):
 
     return data_list
 
-  def revokeCertificate(self, serial, not_after_date, reason=''):
+  def revokeCertificate(self, serial, reason=''):
     """
     Add serial to the list of revoked certificates.
     Associated certificate must expire at (or before) not_after_date, so
@@ -341,11 +341,16 @@ class Storage(object):
     if not cert:
       raise NotFound('No certficate with serial %r' % (serial, ))
 
+    expire_in = cert.expire_after  - datetime.utcnow()
+    if expire_in.days < 0:
+      raise ExpiredCertificate("Certificate with serial %r has expired" \
+              " since %r day(s)." % (serial, -1*expire_in.days))
+
     revoke = Revocation(
         serial=serial,
         creation_date=datetime.utcnow(),
         reason=reason,
-        crt_expire_after=not_after_date
+        crt_expire_after=cert.expire_after
       )
     # Set latest CRL as expired, it will be regenerated
     crl = CertificateRevocationList.query.filter(

caucase/test/testcaweb.py → caucase/test/testweb.py

@@ -27,7 +27,6 @@ from caucase.web import parseArguments, configure_flask
 from OpenSSL import crypto, SSL
 from caucase.exceptions import (NoStorage, NotFound, Found)
 from caucase import utils
-from caucase import db, app
 from flask_testing import TestCase
 from flask import url_for
 
@@ -38,14 +37,16 @@ class CertificateAuthorityWebTest(TestCase):
     configure_flask(parseArguments(['--ca-dir', self.ca_dir, '-s', '/CN=CA Auth Test/emailAddress=xx@example.com']))
 
   def tearDown(self):
-    db.session.remove()
-    db.drop_all()
+    self.db.session.remove()
+    self.db.drop_all()
     if os.path.exists(self.ca_dir):
       shutil.rmtree(self.ca_dir)
 
   def create_app(self):
+    from caucase import db, app
     app.config['TESTING'] = True
     app.config['LIVESERVER_PORT'] = 0
+    self.db = db
     return app
 
   def generateCSR(self, cn="toto.example.com"):

caucase/web.py

@@ -38,8 +38,7 @@ from caucase.exceptions import (NoStorage, NotFound, Found, BadSignature,
                                 CertificateVerificationError,
                                 ExpiredCertificate)
 from functools import wraps
-from caucase import utils
-from caucase import app, db
+from caucase import utils, app, db
 
 class DisabledStringField(StringField):
   def __call__(self, *args, **kwargs):
@@ -191,7 +190,6 @@ def configure_flask(options):
   app.logger.addHandler(logger)
 
   # Instanciate storage
-  # XXX - check  loaded_crt_life_time here
   from caucase.storage import Storage
   storage = Storage(db,
                     max_csr_amount=options.max_request_amount,
@@ -221,7 +219,7 @@ def configure_flask(options):
 def check_authentication(username, password):
   user = app.config.storage.findUser(username)
   if user:
-    return app.user_manager.hash_password(password) == user.password
+    return app.user_manager.verify_password(password, user)
   else:
     return False
 
@@ -234,7 +232,7 @@ def authenticated_method(func):
       auth = request.authorization
       if not auth:
         return abort(401)
-      elif not Users.check_authentication(auth.username, auth.password):
+      elif not check_authentication(auth.username, auth.password):
         return abort(401)
       # Call the actual view
       return func(*args, **kwargs)
@@ -360,12 +358,18 @@ def before_request():
 
 @app.route('/crl', methods=['GET'])
 def get_crl():
+  """
+    Get the lastest CRL (certificate revocation list)
+  """
 
   crl_content = app.config.ca.getCertificateRevocationList()
   return send_file_content(crl_content, 'ca.crl.pem')
 
 @app.route('/csr/<string:csr_id>', methods=['GET'])
 def get_csr(csr_id):
+  """
+    Get a CSR string in PEM format from identified by `csr_id`.
+  """
 
   try:
     csr_content = app.config.ca.getPendingCertificateRequest(csr_id)
@@ -377,6 +381,9 @@ def get_csr(csr_id):
 
 @app.route('/csr', methods=['PUT'])
 def request_cert():
+  """
+    Store certificate signature request (csr) in PEM format
+  """
   csr_content = request.form.get('csr', '').encode('utf-8')
   if not csr_content:
     raise FlaskException("'csr' parameter is mandatory",
@@ -399,6 +406,9 @@ def request_cert():
 @app.route('/csr/<string:csr_id>', methods=['DELETE'])
 @authenticated_method
 def remove_csr(csr_id):
+  """
+    Delete a Certificate signature request. Authentication required
+  """
 
   try:
     app.config.ca.deletePendingCertificateRequest(csr_id)
@@ -412,6 +422,9 @@ def remove_csr(csr_id):
 
 @app.route('/crt/<string:cert_id>', methods=['GET'])
 def get_crt(cert_id):
+  """
+    Get a certificate by the id `cert_id`
+  """
 
   try:
     cert_content = app.config.ca.getCertificate(cert_id)
@@ -421,8 +434,24 @@ def get_crt(cert_id):
 
   return send_file_content(cert_content, cert_id)
 
+@app.route('/crt/serial/<string:serial>', methods=['GET'])
+def crt_fromserial(serial):
+  """
+    Get a certificate by the serial
+  """
+  try:
+    cert_content = app.config.ca.getCertificateFromSerial(serial)
+  except NotFound, e:
+    raise FlaskException("%s" % str(e),
+                status_code=404, payload={"name": "FileNotFound", "code": 1})
+
+  return send_file_content(cert_content, '%s.crt.pem' % serial)
+
 @app.route('/crt/ca.crt.pem', methods=['GET'])
 def get_cacert():
+  """
+    Get CA Certificate in PEM format string.
+  """
 
   ca_cert = app.config.ca.getCACertificate()
 
@@ -430,7 +459,11 @@ def get_cacert():
 
 @app.route('/crt/ca.crt.json', methods=['GET'])
 def get_cacert_json():
-
+  """
+    Return CA certificate chain list, if the CA certificate is being renewed
+    the list will contain the next certificate and the old certificate which
+    will expire soon.
+  """
   ca_chain_list = app.config.ca.getValidCACertificateChain()
 
   return jsonify(ca_chain_list)
@@ -460,6 +493,9 @@ def signcert(csr_key, redirect_to=''):
 @app.route('/crt', methods=['PUT'])
 @authenticated_method
 def sign_cert():
+  """
+    Sign a certificate, require authentication
+  """
   key = request.form.get('csr_id', '').encode('utf-8')
   if not key:
     raise FlaskException("'csr_id' parameter is a mandatory parameter",
@@ -570,9 +606,30 @@ def request_revoke_crt():
   response = Response("", status=201, )
   return response
 
+@app.route('/crt/revoke/serial', methods=['PUT'])
+@authenticated_method
+def revoke_crt():
+  """
+    Directly revoke a certificate from his serial
+  """
 
+  try:
+    serial = request.form.get('serial', '')
+    app.config.ca.revokeCertificateFromSerial(serial)
+  except ValueError, e:
+    traceback.print_exc()
+    raise FlaskException(str(e),
+                payload={"name": "FileFormat", "code": 3})
+  except ExpiredCertificate, e:
+    raise FlaskException(str(e),
+                payload={"name": "FileFormat", "code": 3})
+  except NotFound, e:
+    raise FlaskException(str(e),
+                status_code=404, payload={"name": "FileNotFound", "code": 1})
 
+  response = Response("", status=201)
 
+  return response
 
 
 #Manage routes (Authentication required) - Flask APP

caucase/wsgi.py

@@ -18,7 +18,8 @@
 
 
 import os
-from caucase.web import parseArguments, configure_flask, app
+from caucase import app
+from caucase.web import parseArguments, configure_flask
 from werkzeug.contrib.fixers import ProxyFix
 
 def readConfigFromFile(config_file):
@@ -52,4 +53,4 @@ def start_wsgi():
   app.logger.info("Certificate Authority server ready...")
 
 if __name__ == 'caucase.wsgi':
-  start_wsgi()
\ No newline at end of file
+  start_wsgi()