support a simple shared secret authentication scheme

autoca/ca_app.py

@@ -24,6 +24,16 @@ def content_type(ctype):
     return _ctype_decorator
 
 
+def auth(fn):
+    @functools.wraps(fn)
+    def _auth_wrapper(*args, **kwargs):
+        secret = current_app.config.get('SHARED_SECRET')
+        if secret and request.headers.get('X-Shared-Secret') != secret:
+            return make_response('Unauthorized', 401)
+        return fn(*args, **kwargs)
+    return _auth_wrapper
+
+
 @ca_app.before_request
 def set_ca_wrapper():
     g.ca = current_app.ca
@@ -32,31 +42,32 @@ def set_ca_wrapper():
 @ca_app.route('/ca.pem')
 @content_type('application/x-x509-ca-cert')
 def get_ca():
-    return g.ca.public_ca_pem
+    return g.ca.get_ca()
 
 
 @ca_app.route('/crl.pem')
 @content_type('application/x-x509-ca-cert')
 def get_crl_pem():
-    return g.ca.crl_data_pem
+    return g.ca.get_crl(format='pem')
 
 
 @ca_app.route('/ca.crl')
 @content_type('application/x-pkcs7-crl')
 def get_crl_der():
-    return g.ca.crl_data_der
+    return g.ca.get_crl(format='der')
 
 
 @ca_app.route('/get/<cn>')
 @content_type('application/x-x509-user-cert')
 def get_certificate(cn):
-    cert = g.ca.get_certificate(cn)
+    cert = g.ca.get_certificate(cn, raw=True)
     if not cert:
         abort(404)
-    return crypto.dump_certificate(crypto.FILETYPE_PEM, cert)
+    return cert
 
 
 @ca_app.route('/revoke/<cn>', methods=['POST'])
+@auth
 def revoke(cn):
     g.ca.revoke_certificate(cn)
     return 'ok'
@@ -64,6 +75,7 @@ def revoke(cn):
 
 @ca_app.route('/sign', methods=['POST'])
 @content_type('application/x-x509-user-cert')
+@auth
 def sign():
     if not request.form.get('csr'):
         abort(400)

autoca/ca_stub.py

@@ -15,8 +15,9 @@ class Error(Exception):
 
 class CaStub(object):
 
-    def __init__(self, url):
+    def __init__(self, url, secret=None):
         self.url = url.rstrip('/')
+        self.secret = secret
         self.ca_pem = None
         self._cache_lock = threading.Lock()
 
@@ -27,7 +28,10 @@ class CaStub(object):
                 path = '%s?%s' % (path, urllib.urlencode(args))
             else:
                 data = urllib.urlencode(args)
-        request = urllib2.Request(self.url + path, data)
+        headers = {}
+        if self.secret:
+            headers['X-Shared-Secret'] = self.secret
+        request = urllib2.Request(self.url + path, data, headers)
         try:
             response = urllib2.urlopen(request)
             response_data = response.read()