IntenseWebs/freeipa
3
0
Fork 0
mirror of https://salsa.debian.org/freeipa-team/freeipa.git synced 2024-12-23 15:40:01 -06:00
Code Issues Packages Projects Releases Wiki Activity
3ea044fb59
freeipa/ipalib/pkcs10.py
Rob Crittenden 3ea044fb59 Handle CSRs whether they have NEW in the header or not 
Also consolidate some duplicate code
2010-05-03 17:58:08 -06:00

442 lines
19 KiB
Python
Raw Blame History

# Authors:
# Rob Crittenden <rcritten@redhat.com>
#
# Copyright (C) 2009 Red Hat
# see file 'COPYING' for use and warranty information
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License as
# published by the Free Software Foundation; version 2 only
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
# Read PKCS#10 certificate requests (see RFC 2986 and 5280)
# NOTE: Not every extension is currently handled. Known to now work:
# 2.5.29.37 - extKeyUsage
import sys, string, base64
from pyasn1.type import base,tag,namedtype,namedval,univ,constraint,char,useful
from pyasn1.codec.der import decoder, encoder
from pyasn1 import error
import copy
# Common OIDs found in a subject
oidtable = { "2.5.4.3": "CN",
"2.5.4.6": "C",
"2.5.4.7": "L",
"2.5.4.8": "ST",
"2.5.4.10": "O",
"2.5.4.11": "OU",
"1.2.840.113549.1.9.1": "E",
"0.9.2342.19200300.100.1.25": "DC",
}
# Some useful OIDs
FRIENDLYNAME = '1.2.840.113549.1.9.20'
EXTENSIONREQUEST = '1.2.840.113549.1.9.14'
MAX = 32 # from mozilla/security/nss/lib/util/secasn1t.h
class DirectoryString(univ.Choice):
componentType = namedtype.NamedTypes(
namedtype.NamedType('teletexString', char.TeletexString().subtype(subtypeSpec=constraint.ValueSizeConstraint(1, MAX))),
namedtype.NamedType('printableString', char.PrintableString().subtype(subtypeSpec=constraint.ValueSizeConstraint(1, MAX))),
namedtype.NamedType('universalString', char.UniversalString().subtype(subtypeSpec=constraint.ValueSizeConstraint(1, MAX))),
namedtype.NamedType('utf8String', char.UTF8String().subtype(subtypeSpec=constraint.ValueSizeConstraint(1, MAX))),
namedtype.NamedType('bmpString', char.BMPString().subtype(subtypeSpec=constraint.ValueSizeConstraint(1, MAX))),
)
class AttributeValue(DirectoryString): pass
class AttributeType(univ.ObjectIdentifier): pass
class AttributeTypeAndValue(univ.Sequence):
componentType = namedtype.NamedTypes(
namedtype.NamedType('type', AttributeType()),
namedtype.NamedType('value', AttributeValue()) # FIXME, could be any type
)
class KeyPurposeId(univ.ObjectIdentifier): pass
class ExtKeyUsageSyntax(univ.SequenceOf):
componentType = KeyPurposeId()
class UPN(char.UTF8String):
tagSet = char.UTF8String.tagSet.tagExplicitly(
tag.Tag(tag.tagClassContext, tag.tagFormatConstructed, 0)
)
class AttributeValueSet(univ.SetOf):
componentType = univ.Any()
sizeSpec = univ.SetOf.sizeSpec + constraint.ValueSizeConstraint(1, MAX)
class Attribute(univ.Sequence):
componentType = namedtype.NamedTypes(
namedtype.NamedType('type', AttributeType()),
namedtype.NamedType('values', AttributeValueSet()),
)
class Attributes(univ.SetOf):
componentType = Attribute()
class RelativeDistinguishedName(univ.SetOf):
componentType = AttributeTypeAndValue()
class RDNSequence(univ.SequenceOf):
componentType = RelativeDistinguishedName()
class Name(univ.Choice):
componentType = namedtype.NamedTypes(
namedtype.NamedType('', RDNSequence())
)
def get_components(self):
components = self.getComponentByPosition(0)
complist = []
for idx in range(len(components)):
attrandvalue = components[idx].getComponentByPosition(0)
oid = attrandvalue.getComponentByPosition(0)
# FIXME, should handle any string type
value = attrandvalue.getComponentByPosition(1).getComponentByType(char.PrintableString.tagSet)
if value is None:
value = attrandvalue.getComponentByPosition(1).getComponentByType(char.UTF8String.tagSet)
if value is None:
value = attrandvalue.getComponentByPosition(1).getComponentByType(char.IA5String.tagSet)
vout = value.prettyOut(value).decode('utf-8')
oidout = oid.prettyOut(oid).decode('utf-8')
c = ((oidtable.get(oidout, oidout), vout))
complist.append(c)
return tuple(complist)
class AnotherName(univ.Sequence):
componentType = namedtype.NamedTypes(
namedtype.NamedType('type-id', univ.ObjectIdentifier()),
namedtype.NamedType('value', univ.Any())
)
class rfc822Name(char.IA5String):
tagSet = char.IA5String.tagSet.tagImplicitly(
tag.Tag(tag.tagClassContext, tag.tagFormatSimple, 1)
)
class dNSName(char.IA5String):
tagSet = char.IA5String.tagSet.tagImplicitly(
tag.Tag(tag.tagClassContext, tag.tagFormatSimple, 2)
)
class x400Address(univ.OctetString):
tagSet = univ.OctetString.tagSet.tagImplicitly(
tag.Tag(tag.tagClassUniversal, tag.tagFormatSimple, 3)
)
class directoryName(Name):
tagSet = Name.tagSet.tagImplicitly(
tag.Tag(tag.tagClassUniversal, tag.tagFormatSimple, 4)
)
class uniformResourceIdentifier(char.IA5String):
tagSet = char.IA5String.tagSet.tagImplicitly(
tag.Tag(tag.tagClassContext, tag.tagFormatSimple, 6)
)
# Not all general types are handled, nor are these necessarily done
# per the specification.
class GeneralName(univ.Choice):
componentType = namedtype.NamedTypes(
namedtype.NamedType('otherName', AnotherName().subtype(implicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatSimple, 0))),
namedtype.NamedType('rfc822Name', rfc822Name()), #1
namedtype.NamedType('dNSName', dNSName()), #2
namedtype.NamedType('x400Address', x400Address()), #3
namedtype.NamedType('directoryName', directoryName()), #4
# 5 FIXME
namedtype.NamedType('uniformResourceIdentifier', uniformResourceIdentifier()), #6
# namedtype.NamedType('uniformResourceIdentifier', char.IA5String(tagSet=char.IA5String.tagSet.tagImplicitly(tag.Tag(tag.tagClassUniversal, tag.tagFormatSimple, 6)))),
)
class GeneralNames(univ.SequenceOf):
componentType = GeneralName()
sizeSpec = univ.SequenceOf.sizeSpec + constraint.ValueSizeConstraint(1, MAX)
class SubjectAltName(univ.SequenceOf):
componentType = GeneralName()
sizeSpec = univ.SequenceOf.sizeSpec + constraint.ValueSizeConstraint(1, MAX)
class DistributionPointName(univ.Choice):
componentType = namedtype.NamedTypes(
namedtype.NamedType('fullName', GeneralNames().subtype(implicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatSimple, 0))),
namedtype.NamedType('nameRelativeToCRLIssuer', RelativeDistinguishedName().subtype(implicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatSimple, 1))),
)
class DistributionPoint(univ.Sequence):
componentType = namedtype.NamedTypes(
namedtype.OptionalNamedType('distributionPoint', DistributionPointName().subtype(implicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatSimple, 0))),
namedtype.OptionalNamedType('reasons', univ.BitString().subtype(implicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatSimple, 1))), # FIXME
namedtype.OptionalNamedType('cRLIssuer', GeneralNames().subtype(implicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatSimple, 2))),
)
class cRLDistributionPoints(univ.SequenceOf):
componentType = DistributionPoint()
sizeSpec = univ.SequenceOf.sizeSpec + constraint.ValueSizeConstraint(1, MAX)
class basicConstraints(univ.Sequence):
componentType = namedtype.NamedTypes(
namedtype.DefaultedNamedType('cA', univ.Boolean('False')),
namedtype.OptionalNamedType('pathLenConstraint', univ.Integer()),
)
class AlgorithmIdentifier(univ.Sequence):
componentType = namedtype.NamedTypes(
namedtype.NamedType('algorithm', univ.ObjectIdentifier()),
namedtype.OptionalNamedType('parameters', univ.Any())
)
class SubjectPublicKeyInfo(univ.Sequence):
componentType = namedtype.NamedTypes(
namedtype.NamedType('algorithm', AlgorithmIdentifier()),
namedtype.NamedType('subjectPublicKey', univ.BitString())
)
class Version(univ.Integer): pass
class CertificationRequestInfo(univ.Sequence):
componentType = namedtype.NamedTypes(
namedtype.NamedType('version', Version()),
namedtype.NamedType('subject', Name()),
namedtype.NamedType('subjectPublicKeyInfo', SubjectPublicKeyInfo()),
namedtype.OptionalNamedType('attributes', Attributes().subtype(implicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatSimple, 0)))
)
class CertificationRequest(univ.Sequence):
componentType = namedtype.NamedTypes(
namedtype.NamedType('certificationRequestInfo', CertificationRequestInfo()),
namedtype.NamedType('signatureAlgorithm', AlgorithmIdentifier()),
namedtype.NamedType('signatureValue', univ.BitString())
)
def get_version(self):
info = self.getComponentByName('certificationRequestInfo')
version = info.getComponentByName('version')
return version._value
def get_subject(self):
info = self.getComponentByName('certificationRequestInfo')
return info.getComponentByName('subject')
def get_subjectaltname(self):
attrs = self.get_attributes()
attrdict = dict(attrs)
if EXTENSIONREQUEST in attrdict:
# Extensions are a 3 position tuple
for ext in attrdict[EXTENSIONREQUEST]:
if ext[0] == '2.5.29.17':
# alt name is in the dNSName position
return ext[2][2]
def get_attributes(self):
info = self.getComponentByName('certificationRequestInfo')
attrs = info.getComponentByName('attributes')
attributes = []
for idx in range(len(attrs)):
atype = attrs[idx].getComponentByPosition(0)
aval = attrs[idx].getComponentByPosition(1)
# The attribute list is of type Any, need to re-encode
aenc = encoder.encode(aval, maxChunkSize=1024)
decoded = decoder.decode(aenc)[0]
oid = atype.prettyOut(atype)
if oid == "1.2.840.113549.1.9.20": # PKCS#9 Friendly Name
value = decoded.getComponentByPosition(0)
t = (oid, value.prettyOut(value).decode('utf-8'))
attributes.append(t)
elif oid == "1.2.840.113549.1.9.14": # PKCS#9 Extension Req
extensions = []
extlist = decoded.getComponentByPosition(0)
for jdx in range(len(extlist)):
ext = extlist.getComponentByPosition(jdx)
# An extension has 3 elements:
# oid
# bool - critical
# value
if len(ext) == 2: # If no critical, default to False
extoid = atype.prettyOut(ext.getComponentByPosition(0))
critical = False
extvalue = ext.getComponentByPosition(1)
else:
extoid = atype.prettyOut(ext.getComponentByPosition(0))
critical = bool(ext.getComponentByPosition(1)._value)
extvalue = ext.getComponentByPosition(2)
if extoid == '2.5.29.19': # basicConstraints
extdecoded = decoder.decode(extvalue._value, asn1Spec=basicConstraints())[0]
ca = bool(extdecoded[0])
if len(extdecoded) == 2: # path length is optional
pathlen = extdecoded[1]._value
else:
pathlen = None
constraint = (ca, pathlen)
e = (extoid, critical, constraint)
extensions.append(e)
continue
elif extoid == '2.5.29.31': # cRLDistributionPoints
extdecoded = decoder.decode(extvalue._value, asn1Spec=cRLDistributionPoints())[0]
distpoints = []
for elem in range(len(extdecoded)):
name = extdecoded[elem]
# DistributionPoint is position 0
distpoint = name.getComponentByPosition(0)
# fullName is position 0
fullname = distpoint.getComponentByPosition(0)
for crl in range(len(fullname)):
# Get the GeneralName, URI type
uri = fullname.getComponentByPosition(crl).getComponentByPosition(5)
distpoints.append(uri.prettyOut(uri).decode('utf-8'))
e = (extoid, critical, tuple(distpoints))
extensions.append(e)
continue
# The data is is encoded as "Any". Pull the raw data out
# and re-decode it using a different specification.
try:
extdecoded = decoder.decode(extvalue._value, asn1Spec=GeneralNames())[0]
except error.PyAsn1Error:
# I've seen CSRs where this isn't a sequence of names
# but is a single name, try to handle that too.
try:
extdecoded = decoder.decode(extvalue._value, asn1Spec=GeneralName())[0]
extdecoded = [extdecoded]
except error.PyAsn1Error, e:
# skip for now
generalnames = 9*["Error"]
e = (extoid, critical, tuple(generalnames))
extensions.append(e)
continue
# We now have a list of extensions in the order they
# are in the request as GeneralNames. We iterate through
# each of those to get a GeneralName. We then have to
# iterate through that to find the position set in it.
# Note that not every type will be returned. Those that
# are handled are returned in a tuple in the position
# which they are in the request.
generalnames = 9*[None]
for elem in range(len(extdecoded)):
name = extdecoded[elem]
for n in range(len(name)):
if name[n] is None:
continue
if generalnames[n] is None:
generalnames[n] = []
if n == 3: # OctetString
generalnames[n].append(name[n]._value)
if n in [1, 2, 6]: # IA5String
if n == 6 and extoid == "2.5.29.37":
# Extended key usage
v = copy.deepcopy(extvalue._value)
othername = decoder.decode(v, asn1Spec=ExtKeyUsageSyntax())[0]
keyusage = []
for l in range(len(othername)):
keyusage.append(othername[l].prettyOut(othername[l]))
generalnames[n] = tuple(keyusage)
else:
generalnames[n].append(name[n].prettyOut(name[n]).decode('utf-8'))
if n == 0: # AnotherName
nameoid = name[n].getComponentByPosition(0)
nameoid = nameoid.prettyOut(nameoid)
val = name[n].getComponentByPosition(1)
if nameoid == "1.3.6.1.4.1.311.20.2.3": # UPN
v = copy.deepcopy(val._value)
othername = decoder.decode(v, asn1Spec=UPN())[0]
generalnames[0].append(othername.prettyOut(othername).decode('utf-8'))
e = (extoid, critical, tuple(generalnames))
extensions.append(e)
t = (oid, tuple(extensions))
attributes.append(t)
return tuple(attributes)
def strip_header(csr):
"""
Remove the header and footer from a CSR.
"""
headerlen = 40
s = csr.find("-----BEGIN NEW CERTIFICATE REQUEST-----")
if s == -1:
headerlen = 36
s = csr.find("-----BEGIN CERTIFICATE REQUEST-----")
if s >= 0:
e = csr.find("-----END")
csr = csr[s+headerlen:e]
return csr
def load_certificate_request(csr):
"""
Given a base64-encoded certificate request, with or without the
header/footer, return a request object.
"""
csr = strip_header(csr)
substrate = base64.b64decode(csr)
return decoder.decode(substrate, asn1Spec=CertificationRequest())[0]
if __name__ == '__main__':
# Read PEM certs from stdin and print them out in plain text
stSpam, stHam, stDump = 0, 1, 2
state = stSpam
for certLine in sys.stdin.readlines():
certLine = string.strip(certLine)
if state == stSpam:
if state == stSpam:
if certLine == '-----BEGIN NEW CERTIFICATE REQUEST-----':
certLines = []
state = stHam
continue
if state == stHam:
if certLine == '-----END NEW CERTIFICATE REQUEST-----':
state = stDump
else:
certLines.append(certLine)
complist = []
if state == stDump:
substrate = ''
for certLine in certLines:
substrate = substrate + base64.b64decode(certLine)
request = decoder.decode(substrate, asn1Spec=CertificationRequest())[0]
subject = request.get_subject()
attrs = request.get_attributes()
print "Attributes:"
print attrs
print "Subject:"
complist = subject.get_components()
print complist
out=""
for c in complist:
out = out + "%s=%s," % (c[0], c[1])
print out[:-1]
print request.get_subjectaltname()
# Re-encode the request just to be sure things are working
assert encoder.encode(request, maxChunkSize=1024) == substrate, 'cert recode fails'
state = stSpam
Reference in New Issue View Git Blame Copy Permalink