lauri/certidude
1
0
Fork 0
mirror of https://github.com/laurivosandi/certidude synced 2024-12-22 16:25:17 +00:00
Code Issues Activity

api: Add preliminary SCEP support

Browse Source
This commit is contained in:
Lauri Võsandi 2017-05-18 22:29:49 +03:00
parent a5ad9238a1
commit 5ae872e1ea
7 changed files with 312 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.gitignore vendored
View File

@ -64,3 +64,7 @@ node_modules/
# Ignore autogenerated files
certidude/static/js/nunjucks*
certidude/static/js/templates.js
# Ignore patch
*.orig
*.rej

5
certidude/api/__init__.py
View File

@ -210,6 +210,11 @@ def certidude_app(log_handlers=[]):
# Bootstrap resource
app.add_route("/api/bootstrap/", BootstrapResource())
# Add SCEP handler if we have any whitelisted subnets
if config.SCEP_SUBNETS:
from .scep import SCEPResource
app.add_route("/api/scep/", SCEPResource())
# Add sink for serving static files
app.add_sink(StaticResource(os.path.join(__file__, "..", "..", "static")))

273
certidude/api/scep.py Normal file
View File

@ -0,0 +1,273 @@
from __future__ import unicode_literals, division, absolute_import, print_function
import click
import hashlib
import os
from asn1crypto import cms, algos, x509
from asn1crypto.core import ObjectIdentifier, SetOf, PrintableString
from base64 import b64decode, b64encode
from certbuilder import pem_armor_certificate
from certidude import authority, push, config
from certidude.firewall import whitelist_subnets
from oscrypto import keys, asymmetric, symmetric
from oscrypto.errors import SignatureError
# Monkey patch asn1crypto
class SetOfPrintableString(SetOf):
_child_spec = PrintableString
cms.CMSAttributeType._map['2.16.840.1.113733.1.9.2'] = "message_type"
cms.CMSAttributeType._map['2.16.840.1.113733.1.9.3'] = "pki_status"
cms.CMSAttributeType._map['2.16.840.1.113733.1.9.4'] = "fail_info"
cms.CMSAttributeType._map['2.16.840.1.113733.1.9.5'] = "sender_nonce"
cms.CMSAttributeType._map['2.16.840.1.113733.1.9.6'] = "recipient_nonce"
cms.CMSAttributeType._map['2.16.840.1.113733.1.9.7'] = "trans_id"
cms.CMSAttribute._oid_specs['message_type'] = SetOfPrintableString
cms.CMSAttribute._oid_specs['pki_status'] = SetOfPrintableString
cms.CMSAttribute._oid_specs['fail_info'] = SetOfPrintableString
cms.CMSAttribute._oid_specs['sender_nonce'] = cms.SetOfOctetString
cms.CMSAttribute._oid_specs['recipient_nonce'] = cms.SetOfOctetString
cms.CMSAttribute._oid_specs['trans_id'] = SetOfPrintableString
class SCEPError(Exception): code = 25 # system failure
class SCEPBadAlg(SCEPError): code = 0
class SCEPBadMessageCheck(SCEPError): code = 1
class SCEPBadRequest(SCEPError): code = 2
class SCEPBadTime(SCEPError): code = 3
class SCEPBadCertId(SCEPError): code = 4
class SCEPResource(object):
@whitelist_subnets(config.SCEP_SUBNETS)
def on_get(self, req, resp):
operation = req.get_param("operation")
if operation.lower() == "getcacert":
resp.stream = keys.parse_certificate(authority.ca_buf).dump()
resp.append_header("Content-Type", "application/x-x509-ca-cert")
return
# Parse CA certificate
fh = open(config.AUTHORITY_CERTIFICATE_PATH)
server_certificate = asymmetric.load_certificate(fh.read())
fh.close()
# If we bump into exceptions later
encrypted_container = b""
attr_list = [
cms.CMSAttribute({
'type': "message_type",
'values': ["3"]
}),
cms.CMSAttribute({
'type': "pki_status",
'values': ["2"] # rejected
})
]
try:
info = cms.ContentInfo.load(b64decode(req.get_param("message", required=True)))
###############################################
### Verify signature of the outer container ###
###############################################
signed_envelope = info['content']
encap_content_info = signed_envelope['encap_content_info']
encap_content = encap_content_info['content']
# TODO: try except
current_certificate, = signed_envelope["certificates"]
signer, = signed_envelope["signer_infos"]
# TODO: compare cert to current one if we are renewing
assert signer["digest_algorithm"]["algorithm"].native == "md5"
assert signer["signature_algorithm"]["algorithm"].native == "rsassa_pkcs1v15"
message_digest = None
transaction_id = None
sender_nonce = None
for attr in signer["signed_attrs"]:
if attr["type"].native == "sender_nonce":
sender_nonce, = attr["values"]
elif attr["type"].native == "trans_id":
transaction_id, = attr["values"]
elif attr["type"].native == "message_digest":
message_digest, = attr["values"]
if hashlib.md5(encap_content.native).digest() != message_digest.native:
raise SCEPBadMessageCheck()
assert message_digest
msg = signer["signed_attrs"].dump(force=True)
assert msg[0] == b"\xa0", repr(msg[0])
# Verify signature
try:
asymmetric.rsa_pkcs1v15_verify(
asymmetric.load_certificate(current_certificate.dump()),
signer["signature"].native,
b"\x31" + msg[1:], # wtf?!
"md5")
except SignatureError:
raise SCEPBadMessageCheck()
###############################
### Decrypt inner container ###
###############################
info = cms.ContentInfo.load(encap_content.native)
encrypted_envelope = info['content']
encrypted_content_info = encrypted_envelope['encrypted_content_info']
iv = encrypted_content_info['content_encryption_algorithm']['parameters'].native
if encrypted_content_info['content_encryption_algorithm']["algorithm"].native != "des":
raise SCEPBadAlgo()
encrypted_content = encrypted_content_info['encrypted_content'].native
recipient, = encrypted_envelope['recipient_infos']
if recipient.native["rid"]["serial_number"] != authority.ca_cert.serial:
raise SCEPBadCertId()
# Since CA private key is not directly readable here, we'll redirect it to signer socket
key = b64decode(authority.signer_exec("decrypt-pkcs7", b64encode(recipient.native["encrypted_key"])))
if len(key) == 8: key = key * 3 # Convert DES to 3DES
buf = symmetric.tripledes_cbc_pkcs5_decrypt(key, encrypted_content, iv)
_, common_name = authority.store_request(buf, overwrite=True)
cert, buf = authority.sign(common_name, overwrite=True)
signed_certificate = asymmetric.load_certificate(buf)
content = signed_certificate.asn1.dump()
except SCEPError, e:
attr_list.append(cms.CMSAttribute({
'type': "fail_info",
'values': ["%d" % e.code]
}))
else:
##################################
### Degenerate inner container ###
##################################
degenerate = cms.ContentInfo({
'content_type': 'signed_data',
'content': cms.SignedData({
'version': 'v1',
'certificates': [signed_certificate.asn1],
'digest_algorithms': [cms.DigestAlgorithm({
'algorithm':'md5'
})],
'encap_content_info': {
'content_type': 'data',
'content': cms.ContentInfo({
'content_type': 'signed_data',
'content': None
}).dump()
},
'signer_infos': []
})
})
################################
### Encrypt middle container ###
################################
key = os.urandom(8)
iv, encrypted_content = symmetric.des_cbc_pkcs5_encrypt(key, degenerate.dump(), os.urandom(8))
assert degenerate.dump() == symmetric.tripledes_cbc_pkcs5_decrypt(key*3, encrypted_content, iv)
ri = cms.RecipientInfo({
'ktri': cms.KeyTransRecipientInfo({
'version': 'v0',
'rid': cms.RecipientIdentifier({
'issuer_and_serial_number': cms.IssuerAndSerialNumber({
'issuer': current_certificate.chosen["tbs_certificate"]["issuer"],
'serial_number': current_certificate.chosen["tbs_certificate"]["serial_number"],
}),
}),
'key_encryption_algorithm': {
'algorithm': 'rsa'
},
'encrypted_key': asymmetric.rsa_pkcs1v15_encrypt(
asymmetric.load_certificate(current_certificate.chosen.dump()), key)
})
})
encrypted_container = cms.ContentInfo({
'content_type': 'enveloped_data',
'content': cms.EnvelopedData({
'version': 'v1',
'recipient_infos': [ri],
'encrypted_content_info': {
'content_type': 'data',
'content_encryption_algorithm': {
'algorithm': 'des',
'parameters': iv
},
'encrypted_content': encrypted_content
}
})
}).dump()
attr_list = [
cms.CMSAttribute({
'type': 'message_digest',
'values': [hashlib.sha1(encrypted_container).digest()]
}),
cms.CMSAttribute({
'type': "message_type",
'values': ["3"]
}),
cms.CMSAttribute({
'type': "pki_status",
'values': ["0"] # ok
})
]
finally:
##############################
### Signed outer container ###
##############################
attrs = cms.CMSAttributes(attr_list + [
cms.CMSAttribute({
'type': "recipient_nonce",
'values': [sender_nonce]
}),
cms.CMSAttribute({
'type': 'trans_id',
'values': [transaction_id]
})
])
signer = cms.SignerInfo({
"signed_attrs": attrs,
'version':'v1',
'sid': cms.SignerIdentifier({
'issuer_and_serial_number': cms.IssuerAndSerialNumber({
'issuer': server_certificate.asn1["tbs_certificate"]["issuer"],
'serial_number': server_certificate.asn1["tbs_certificate"]["serial_number"],
}),
}),
'digest_algorithm': algos.DigestAlgorithm({'algorithm': 'sha1'}),
'signature_algorithm': algos.SignedDigestAlgorithm({'algorithm': 'rsassa_pkcs1v15'}),
'signature': b64decode(authority.signer_exec("sign-pkcs7", b64encode(b"\x31" + attrs.dump()[1:])))
})
resp.append_header("Content-Type", "application/x-pki-message")
resp.body = cms.ContentInfo({
'content_type': 'signed_data',
'content': cms.SignedData({
'version': 'v1',
'certificates': [x509.Certificate.load(server_certificate.asn1.dump())], # wat
'digest_algorithms': [cms.DigestAlgorithm({
'algorithm':'sha1'
})],
'encap_content_info': {
'content_type': 'data',
'content': encrypted_container
},
'signer_infos': [signer]
})
}).dump()

13
certidude/authority.py
View File

@ -12,6 +12,7 @@ from cryptography import x509
from cryptography.x509.oid import NameOID, ExtensionOID, ExtendedKeyUsageOID
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.serialization import Encoding
from certidude import config, push, mailer, const
from certidude import errors
from jinja2 import Template
@ -81,7 +82,13 @@ def store_request(buf, overwrite=False):
if not buf:
raise ValueError("No signing request supplied")
csr = x509.load_pem_x509_csr(buf, backend=default_backend())
if isinstance(buf, unicode):
csr = x509.load_pem_x509_csr(buf, backend=default_backend())
elif isinstance(buf, str):
csr = x509.load_der_x509_csr(buf, backend=default_backend())
buf = csr.public_bytes(Encoding.PEM)
else:
raise ValueError("Invalid type, expected str for PEM and bytes for DER")
common_name, = csr.subject.get_attributes_for_oid(NameOID.COMMON_NAME)
# TODO: validate common name again
@ -92,7 +99,7 @@ def store_request(buf, overwrite=False):
# If there is cert, check if it's the same
if os.path.exists(request_path):
if os.path.exists(request_path) and not overwrite:
if open(request_path).read() == buf:
raise errors.RequestExists("Request already exists")
else:
@ -106,7 +113,7 @@ def store_request(buf, overwrite=False):
mailer.send("request-stored.md",
attachments=(attach_csr,),
common_name=common_name.value)
return csr
return csr, common_name.value
def signer_exec(cmd, *bits):

2
certidude/config.py
View File

@ -32,6 +32,8 @@ AUTOSIGN_SUBNETS = set([ipaddress.ip_network(j) for j in
cp.get("authorization", "autosign subnets").split(" ") if j])
REQUEST_SUBNETS = set([ipaddress.ip_network(j) for j in
cp.get("authorization", "request subnets").split(" ") if j]).union(AUTOSIGN_SUBNETS)
SCEP_SUBNETS = set([ipaddress.ip_network(j) for j in
cp.get("authorization", "scep subnets").split(" ") if j])
AUTHORITY_DIR = "/var/lib/certidude"
AUTHORITY_PRIVATE_KEY_PATH = cp.get("authority", "private key path")

16
certidude/signer.py
View File

@ -5,11 +5,13 @@ import socket
import os
import asyncore
import asynchat
from base64 import b64decode, b64encode
from certidude import const, config
from cryptography import x509
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives import hashes, padding, serialization
from cryptography.hazmat.primitives.serialization import Encoding
from cryptography.hazmat.primitives.asymmetric import padding
from datetime import datetime, timedelta
from cryptography.x509.oid import NameOID, ExtendedKeyUsageOID, AuthorityInformationAccessOID
import random
@ -64,6 +66,18 @@ class SignHandler(asynchat.async_chat):
self.close()
raise asyncore.ExitNow()
elif cmd == "sign-pkcs7":
signer = self.server.private_key.signer(
padding.PKCS1v15(),
hashes.SHA1()
)
signer.update(b64decode(body))
self.send(b64encode(signer.finalize()))
elif cmd == "decrypt-pkcs7":
self.send(b64encode(self.server.private_key.decrypt(b64decode(body), padding.PKCS1v15())))
self.close()
elif cmd == "sign-request":
# Only common name and public key are used from request
request = x509.load_pem_x509_csr(body, default_backend())

3
certidude/templates/server/server.conf
View File

@ -56,6 +56,9 @@ request subnets = 0.0.0.0/0
# Certificates are automatically signed for these subnets
autosign subnets = 127.0.0.0/8 10.0.0.0/8 172.16.0.0/12 192.168.0.0/16
# Simple Certificate Enrollment Protocol enabled subnets
scep subnets =
[logging]
# Disable logging
;backend =