Helpers for accessing AKI/SKI extensions of certs/crls

lib/openssl/x509.rb

@@ -60,6 +60,109 @@ def to_h # {"oid"=>sn|ln, "value"=>value, "critical"=>true|false}
  def to_a
  [ self.oid, self.value, self.critical? ]
  end
+
+ module Helpers
+ def find_extension(oid)
+ extensions.find { |e| e.oid == oid }
+ end
+
+ def x509_name_from_general_name(gn_asn1)
+ unless gn_asn1.tag_class == :CONTEXT_SPECIFIC
+ raise ASN1::ASN1Error "invalid GeneralName"
+ end
+
+ if gn_asn1.tag == 4
+ Name.new(gn_asn1.value.first.to_der)
+ else
+ # TODO: raise error instead of returning nil?
+ # this is some other kind of general name.
+ nil
+ end
+ end
+ end
+
+ module SubjectKeyIdentifier
+ include Helpers
+
+ # Described in RFC5280 Section 4.2.1.2
+ #
+ # Returns a binary String key identifier or nil.
+ def subject_key_identifier
+ ext = find_extension("subjectKeyIdentifier")
+ return nil if ext.nil?
+
+ ski_asn1 = ASN1.decode(ext.value_der)
+ if ext.critical? || ski_asn1.tag_class != :UNIVERSAL || ski_asn1.tag != ASN1::OCTET_STRING
+ raise ASN1::ASN1Error "invalid extension"
+ end
+
+ ski_asn1.value
+ end
+ end
+
+ module AuthorityKeyIdentifier
+ include Helpers
+
+ # Described in RFC5280 Section 4.2.1.1
+ #
+ # Returns a Hash with keys :key_identifier, :authority_cert_issuer,
+ # and :authority_cert_serial_number.
+ def authority_key_identifier
+ ext = find_extension("authorityKeyIdentifier")
+ return nil if ext.nil?
+
+ aki_asn1 = ASN1.decode(ext.value_der)
+ if ext.critical? || aki_asn1.tag_class != :UNIVERSAL || aki_asn1.tag != ASN1::SEQUENCE
+ raise ASN1::ASN1Error "invalid extension"
+ end
+
+ fields = {}
+
+ key_id = aki_asn1.value.find do |v|
+ v.tag_class == :CONTEXT_SPECIFIC && v.tag == 0
+ end
+
+ issuer = aki_asn1.value.find do |v|
+ v.tag_class == :CONTEXT_SPECIFIC && v.tag == 1
+ end
+
+ serial = aki_asn1.value.find do |v|
+ v.tag_class == :CONTEXT_SPECIFIC && v.tag == 2
+ end
+
+ # must have neither or both of issuer and serial
+ if (!issuer.nil? && serial.nil?) || (issuer.nil? && !serial.nil?)
+ raise ASN1::ASN1Error "invalid extension"
+ end
+
+ fields[:key_identifier] = if !key_id.nil?
+ key_id.value
+ else
+ nil
+ end
+
+ fields[:authority_cert_issuer] = if !issuer.nil?
+ # TODO raise if there are more than one values? GeneralNames is a
+ # SEQUENCE.
+ x509_name_from_general_name(issuer.value.first)
+ else
+ nil
+ end
+
+ fields[:authority_cert_serial_number] = if !serial.nil?
+ # encode/decode as integer to get value parsed as BN
+ ASN1.decode(ASN1::ASN1Data.new(
+ serial.value,
+ ASN1::INTEGER,
+ :UNIVERSAL
+ ).to_der).value
+ else
+ nil
+ end
+
+ fields
+ end
+ end
  end
 
  class Name
@@ -179,6 +282,9 @@ def cleanup
  end
 
  class Certificate
+ include Extension::SubjectKeyIdentifier
+ include Extension::AuthorityKeyIdentifier
+
  def pretty_print(q)
  q.object_group(self) {
  q.breakable
@@ -192,6 +298,8 @@ def pretty_print(q)
  end
 
  class CRL
+ include Extension::AuthorityKeyIdentifier
+
  def ==(other)
  return false unless CRL === other
  to_der == other.to_der

test/test_x509cert.rb

@@ -73,9 +73,15 @@ def test_extension
  ["basicConstraints","CA:TRUE",true],
  ["keyUsage","keyCertSign, cRLSign",true],
  ["subjectKeyIdentifier","hash",false],
- ["authorityKeyIdentifier","keyid:always",false],
+ ["authorityKeyIdentifier","issuer:always,keyid:always",false],
  ]
  ca_cert = issue_cert(@ca, @rsa2048, 1, ca_exts, nil, nil)
+ aki_fields = ca_cert.authority_key_identifier
+ keyid = get_subject_key_id(ca_cert.to_der, hex: false)
+ assert_equal keyid, aki_fields[:key_identifier]
+ assert_equal ca_cert.subject, aki_fields[:authority_cert_issuer]
+ assert_equal ca_cert.serial, aki_fields[:authority_cert_serial_number]
+ assert_equal keyid, ca_cert.subject_key_identifier
  ca_cert.extensions.each_with_index{|ext, i|
  assert_equal(ca_exts[i].first, ext.oid)
  assert_equal(ca_exts[i].last, ext.critical?)
@@ -84,7 +90,7 @@ def test_extension
  ee1_exts = [
  ["keyUsage","Non Repudiation, Digital Signature, Key Encipherment",true],
  ["subjectKeyIdentifier","hash",false],
- ["authorityKeyIdentifier","keyid:always",false],
+ ["authorityKeyIdentifier","issuer:always,keyid:always",false],
  ["extendedKeyUsage","clientAuth, emailProtection, codeSigning",false],
  ["subjectAltName","email:ee1@ruby-lang.org",false],
  ]
@@ -94,6 +100,10 @@ def test_extension
  assert_equal(ee1_exts[i].first, ext.oid)
  assert_equal(ee1_exts[i].last, ext.critical?)
  }
+
+ no_exts_cert = issue_cert(@ca, @rsa2048, 1, [], nil, nil)
+ assert_equal nil, no_exts_cert.authority_key_identifier
+ assert_equal nil, no_exts_cert.subject_key_identifier
  end
 
  def test_sign_and_verify_rsa_sha1

test/test_x509crl.rb

@@ -118,7 +118,7 @@ def test_extension
  ["keyUsage", "cRLSign, keyCertSign", true],
  ]
  crl_exts = [
- ["authorityKeyIdentifier", "keyid:always", false],
+ ["authorityKeyIdentifier", "issuer:always,keyid:always", false],
  ["issuerAltName", "issuer:copy", false],
  ]
 
@@ -131,6 +131,18 @@ def test_extension
  assert_equal("crlNumber", exts[0].oid)
  assert_equal(false, exts[0].critical?)
 
+ expected_keyid = OpenSSL::TestUtils.get_subject_key_id(cert, hex: false)
+ actual_keyid = crl.authority_key_identifier[:key_identifier]
+ assert_equal expected_keyid, actual_keyid
+
+ expected_issuer = cert.issuer
+ actual_issuer = crl.authority_key_identifier[:authority_cert_issuer]
+ assert_equal expected_issuer, actual_issuer
+
+ expected_serial = cert.serial
+ actual_serial = crl.authority_key_identifier[:authority_cert_serial_number]
+ assert_equal expected_serial, actual_serial
+
  assert_equal("authorityKeyIdentifier", exts[1].oid)
  keyid = OpenSSL::TestUtils.get_subject_key_id(cert)
  assert_match(/^keyid:#{keyid}/, exts[1].value)
@@ -155,6 +167,10 @@ def test_extension
  assert_equal("issuerAltName", exts[2].oid)
  assert_equal("email:xyzzy@ruby-lang.org", exts[2].value)
  assert_equal(false, exts[2].critical?)
+
+ no_ext_crl = issue_crl([], 1, Time.now, Time.now+1600, [],
+ cert, @rsa2048, OpenSSL::Digest::SHA1.new)
+ assert_equal nil, no_ext_crl.authority_key_identifier
  end
 
  def test_crlnumber

test/utils.rb

@@ -111,13 +111,18 @@ def issue_crl(revoke_info, serial, lastup, nextup, extensions,
  crl
  end
 
- def get_subject_key_id(cert)
+ def get_subject_key_id(cert, hex: true)
  asn1_cert = OpenSSL::ASN1.decode(cert)
  tbscert = asn1_cert.value[0]
  pkinfo = tbscert.value[6]
  publickey = pkinfo.value[1]
  pkvalue = publickey.value
- OpenSSL::Digest::SHA1.hexdigest(pkvalue).scan(/../).join(":").upcase
+ digest = OpenSSL::Digest::SHA1.digest(pkvalue)
+ if hex
+ digest.unpack("H2"*20).join(":").upcase
+ else
+ digest
+ end
  end
 
  def openssl?(major = nil, minor = nil, fix = nil, patch = 0)