initial import

.htaccess

+RewriteEngine on
+RewriteBase /
+RewriteRule ^(.*)$ autoca.py/$1 [L]
+

autoca.py

+#!/usr/bin/python
+# A minimal RESTful Certification Authority
+# by <ale@incal.net>, 2010
+# $Id$
+
+import os
+import certutil
+from OpenSSL import crypto
+from bottle import route, run, request, response, abort, send_file
+
+X509_MIME = 'application/x-x509-ca-cert'
+ca_conf_file = os.path.join(os.path.dirname(__file__), "ca.yml")
+ca = certutil.CA(ca_conf_file)
+cwd = os.getcwd()
+
+@route('/sign', method='POST')
+def sign():
+    if 'cert' not in request.POST:
+        abort(404)
+    try:
+        csr_data = request.POST['cert'].value
+        csr = crypto.load_certificate_request(crypto.FILETYPE_PEM, csr_data)
+    except Exception, e:
+        print "Exception while decoding CSR data:", e
+        abort(400)
+    signed_cert = ca.sign_certificate(csr)
+    response.content_type = X509_MIME
+    return crypto.dump_certificate(crypto.FILETYPE_PEM, signed_cert)
+
+@route('/ca.pem')
+def get_ca_cert():
+    return send_file(ca.cacrtpath, cwd, mimetype=X509_MIME)
+
+@route('/get/:cn')
+def get_cert(cn):
+    if cn.endswith('.pem'):
+        cn = cn[:-4]
+    return send_file(ca.path_to_cert(cn), cwd, mimetype=X509_MIME)
+
+if __name__ == "__main__":
+    run()

ca.yml

+basedir: ./ca
+default_subj:
+  emailAddress: certs@autistici.org
+  C: XX
+  ST: XX
+  L: Nowhere
+  O: Autistici/Inventati
+ca_subj:
+  CN: Autistici/Inventati AutoCA

certutil.py

+# $Id$
+
+from OpenSSL import crypto
+import os
+import socket
+import time
+import yaml
+
+DEFAULT_CONFIG_FILE = "/etc/autoca.conf"
+DIGEST = "sha1"
+
+def _writeto(file, contents):
+    fd = open(file, "w")
+    fd.write(contents)
+    fd.close()
+
+def _readfrom(file):
+    fd = open(file, "r")
+    try:
+        return fd.read()
+    finally:
+        fd.close()
+
+
+class CA(object):
+
+    def __init__(self, cf_file=DEFAULT_CONFIG_FILE, load=True):
+        self._parse_config(cf_file)
+        self.basedir = self.cf["basedir"]
+        self.cakeypath = os.path.join(self.basedir, ".private", "ca.key")
+        self.cacrtpath = os.path.join(self.basedir, "ca.pem")
+        if load:
+            self._load_ca_keys()
+
+    def _parse_config(self, file):
+        fd = open(file, "r")
+        self.cf = yaml.load(fd)
+        fd.close()
+
+    def _incr_serial(self):
+        serialpath = os.path.join(self.basedir, "serial")
+        try:
+            fd = open(serialpath, "r+")
+            cur_serial = int(fd.read().strip())
+            cur_serial += 1
+        except IOError:
+            fd = open(serialpath, "w")
+            cur_serial = int(time.time())
+        fd.seek(0)
+        fd.truncate()
+        fd.write("%d\n" % cur_serial)
+        fd.close()
+        return cur_serial
+
+    def _setup_ca(self):
+        if not os.path.exists(self.basedir):
+            os.mkdir(self.basedir, 0700)
+        for dir in (".private", "certs"):
+            fulldir = os.path.join(self.basedir, dir)
+            if not os.path.exists(fulldir):
+                os.mkdir(fulldir, 0700)
+        if not os.path.exists(self.cakeypath) or not os.path.exists(self.cacrtpath):
+            ca_key = self.create_rsa_key_pair(4096)
+            ca_req = self.create_cert_request(ca_key, **(self.cf["ca_subj"]))
+            ca_crt = self._sign_certificate(ca_req, ca_key, ca_req)
+
+            _writeto(self.cakeypath, crypto.dump_privatekey(crypto.FILETYPE_PEM, ca_key))
+            os.chmod(self.cakeypath, 0400)
+            _writeto(self.cacrtpath, crypto.dump_certificate(crypto.FILETYPE_PEM, ca_crt))
+
+    def _load_ca_keys(self):
+        self._setup_ca()
+        self.ca_key = crypto.load_privatekey(
+            crypto.FILETYPE_PEM, _readfrom(self.cakeypath))
+        self.public_ca_pem = _readfrom(self.cacrtpath)
+        self.ca_crt = crypto.load_certificate(
+            crypto.FILETYPE_PEM, self.public_ca_pem)
+
+    def create_rsa_key_pair(self, bits=1024):
+        pkey = crypto.PKey()
+        pkey.generate_key(crypto.TYPE_RSA, bits)
+        return pkey
+
+    def create_cert_request(self, pkey, **attrs):
+        if "CN" not in attrs:
+            attrs["CN"] = socket.gethostname()
+        req = crypto.X509Req()
+        subj = req.get_subject()
+        tattrs = self.cf["default_subj"].copy()
+        tattrs.update(attrs)
+        for key, value in tattrs.items():
+            setattr(subj, key, value)
+        req.set_pubkey(pkey)
+        req.sign(pkey, DIGEST)
+        return req
+
+    def _sign_certificate(self, req, ca_key, ca_crt):
+        cert = crypto.X509()
+        cert.set_serial_number(self._incr_serial())
+        cert.gmtime_adj_notBefore(0)
+        cert.gmtime_adj_notAfter(86400 * 365)
+        cert.set_issuer(ca_crt.get_subject())
+        cert.set_subject(req.get_subject())
+        cert.set_pubkey(req.get_pubkey())
+        cert.sign(ca_key, DIGEST)
+        return cert
+
+    def sign_certificate(self, req):
+        cert = self._sign_certificate(req, self.ca_key, self.ca_crt)
+        local_crt = self.path_to_cert(req.get_subject().CN)
+        _writeto(local_crt,
+                 crypto.dump_certificate(crypto.FILETYPE_PEM, cert))
+        return cert
+
+    def path_to_cert(self, cn):
+        return os.path.join(self.cf["basedir"], "certs", "%s.pem" % cn.lower())

newcert.py

+#!/usr/bin/python
+# An example of how to create a new certificate request with certutil
+# $Id$
+
+import os, sys
+import certutil
+from OpenSSL import crypto
+
+def main():
+    cn = sys.argv[1]
+    ca = certutil.CA("ca.yml", load=False)
+    pkey = ca.create_rsa_key_pair()
+    req = ca.create_cert_request(pkey, CN=cn)
+
+    open("newreq.key", "w", 0600).write(crypto.dump_privatekey(crypto.FILETYPE_PEM, pkey))
+    open("newreq.csr", "w").write(crypto.dump_certificate_request(crypto.FILETYPE_PEM, req))
+    print "key and request written to newreq.{key,csr}"
+
+if __name__ == "__main__":
+    main()