Changeset 868

Timestamp:

07/20/06 13:22:07 (7 years ago)

Author:

mglb1

Message:

• Added support for revoking certificates
• Removed crl/ directory - not required
• Added findByCN function to search for a certificate

File:

• ccsd/trunk/crcnetd/_utils/ccsd_ca.py (modified) (6 diffs)

ccsd/trunk/crcnetd/_utils/ccsd_ca.py

@@ -49 +49 @@
 CERT_PARAM_NAMES = ["C", "ST", "L", "O", "OU", "CN", "emailAddress"]
 
+REVOKE_UNSPECIFIED = "unspecified"
+REVOKE_COMPROMISED = "keyCompromise"
+REVOKE_SUPERSEDED = "superseded"
+REVOKE_OBSOLETE = "cessationOfOperation"
+
 certParams = {}
 
@@ -63 +68 @@
     global certParams
     return certParams
+
+def emptyCertParameters():
+    """Helper function returns a empty set of certificate parameters"""
+    global CERT_PARAM_NAMES
+    params = {}
+    for param in CERT_PARAM_NAMES:
+        params[param] = ""
+    return params
 
 @exportViaXMLRPC(SESSION_RO, AUTH_USER)
@@ -191 +204 @@
             s+=1
             log_info("CA: Certificate storage directory created")
-        if not os.path.exists("%s/crl" % self.rDir):
-            # Storage directory for CRLs needs creating
-            ensureDirExists("%s/crl" % self.rDir)
-            self.mSvn.add("%s/crl" % self.rDir, False)
-            s+=1
-            log_info("CA: CRL storage directory created")
 
         # Check for required files
@@ -245 +252 @@
             return
         
-        if s>0 and s!=5:
+        if s>0 and s!=4:
             # Warn if only partial changes were made
             log_warn("CA: Initialised from incomplete state!")
@@ -545 +552 @@
                     "%s/index.txt.attr" % self.rDir, certfile]
             self.checkin("Signed new certificate for %s" % cn, paths)
+            log_info("CA: Signed new certificate (0x%s) for %s" % (serial, cn))
         finally:
             os.unlink(filename)
@@ -551 +559 @@
         cert = open(certfile, "r").read()
         return cert
+    
+    def revoke(self, serial, reasonCode=REVOKE_UNSPECIFIED, reasonText=""):
+        """Revokes the specified certificate optionally giving a reason
+        
+        This function will revoke the key, giving the reason specified, the
+        certificate will be moved from certs/XX.pem to certs/XX-revoked.pem
+        and a new CRL will be issued. This function will commit changes to 
+        index.txt. To be safe it is best to ensure that there are no other
+        pending changes before calling this method.
+        """
+        
+        # Sign the key
+        os.environ["CCS_CA_DIR"] = self.rDir
+        (fdi, fdo) = os.popen2("openssl ca -config %s/ca.cnf -revoke " \
+                "%s/certs/%s.pem -crl_reason %s 2>&1" % \
+                (self.rDir, self.rDir, serial, reasonCode))
+        fdi.close()
+        lines = fdo.readlines()
+        fdo.close()
+        (fdi, fdo) = os.popen2("openssl ca -config %s/ca.cnf -gencrl -out " \
+                "%s/crl.pem 2>&1" % (self.rDir, self.rDir))
+        fdi.close()
+        rlines = fdo.readlines()
+        fdo.close()
+        del os.environ["CCS_CA_DIR"]
+        
+        # Find the works "Revoking Certificate" in the output
+        try:
+            ok = False
+            for line in lines:
+                if line.strip().startswith("Revoking Certificate"):
+                    ok=True
+                    break
+            if not ok:
+                raise ccs_ca_error()
+        except:
+            log_debug("".join(lines))
+            raise ccs_ca_error("Could not validate certificate revocation!")
+
+        # Move the certificate
+        try:
+            self.mSvn.move("%s/certs/%s.pem" % (self.rDir, serial), \
+                    "%s/certs/%s-revoked.pem" % (self.rDir, serial))
+        except:
+            log_error("Could not move revoked certificate to new name!", \
+                    sys.exc_info())
+
+        # Commit the changes
+        message = "Revoked (%s) certificate 0x%s: %s" % \
+                (reasonCode, serial, reasonText)
+        self.checkin(message)
+        log_info("CA: %s" % message)
+        
+    def findByCN(self, desiredCN):
+        """Searches the certificate database for records with matching CNs"""
+        certs = []
+        
+        # Read the database 
+        try:
+            fd = open("%s/index.txt" % self.rDir, "r")
+            lines = fd.readlines()
+            fd.close()
+        except:
+            raise ccs_ca_error("Unable to read certificate database!")
+
+        # Parse the database
+        n=0
+        for line in lines:
+            n+=1
+            parts = line.split("\t")
+            if len(parts) != 6:
+                log_warn("Skipping malformed line %s in certificate DB" % \
+                        n)
+                continue
+            # Look for the CN
+            if parts[5].find("CN=%s" % desiredCN) == -1:
+                continue
+            # Certificate parameters
+            t = parts[5][1:].split("/")
+            params = emptyCertParameters()
+            for p in t:
+                pp = p.split("=")
+                params[pp[0]] = pp[1]
+            # Store the match
+            cert = {"state":parts[0], "exp_date":parts[1], \
+                    "rev_date":parts[2], "serial":parts[3], "file":parts[4], \
+                    "params":params}
+            # Add to the list
+            certs.append(cert)
+
+        # Return results
+        return certs
 
 #####################################################################