CSR Generator — Ruby source

# CSR Generator — ASN.1 DER encoding + PKCS#10 certificate-request assembly.
#
# Language: Ruby (3.x, standard library only — openssl, base64, ipaddr)
# Source:   CosmoDev polyglot showcase port of the CSR Generator tool, ported
#           from src/lib/csr-generator.ts (the canonical TypeScript
#           implementation).
# License:  display source — part of CosmoDev's polyglot tool pages.
#
# The DER encoder is written out by hand, exactly as the TS reference does on
# Web Crypto: length headers, INTEGER / OID / string primitives, SEQUENCE and
# SET constructors, then the CertificationRequestInfo, the signature, and the
# PEM wrapper. OpenSSL is used only for what it must be — key generation and the
# signature itself.
#
# One deliberate difference from the TS port: Web Crypto returns ECDSA
# signatures as raw r||s (IEEE P1363), so the TS code converts them to the
# ASN.1 ECDSA-Sig-Value that PKCS#10 requires. OpenSSL already emits that DER
# SEQUENCE, so no conversion step exists here.
#
# Nothing here talks to the network: the key pair is generated locally, the CSR
# is signed locally, and both PEM blocks are returned to the caller.

require 'openssl'
require 'base64'
require 'ipaddr'

module CsrGenerator
  Options = Struct.new(
    :common_name,
    :organization,
    :country,
    :state,
    :locality,
    :email,
    :key_algorithm,     # 'RSA-2048' | 'RSA-4096' | 'ECDSA-P256'
    :subject_alt_names, # Array of DNS names / IPs / emails / URIs
    keyword_init: true
  )

  Result = Struct.new(:csr, :private_key, keyword_init: true)

  KEY_ALGORITHMS = %w[RSA-2048 RSA-4096 ECDSA-P256].freeze

  # --- object identifiers -----------------------------------------------------

  OID_COUNTRY          = '2.5.4.6'
  OID_STATE            = '2.5.4.8'
  OID_LOCALITY         = '2.5.4.7'
  OID_ORGANIZATION     = '2.5.4.10'
  OID_COMMON_NAME      = '2.5.4.3'
  OID_EMAIL            = '1.2.840.113549.1.9.1'
  OID_EXT_REQUEST      = '1.2.840.113549.1.9.14' # pkcs-9 at extensionRequest
  OID_SUBJECT_ALT_NAME = '2.5.29.17'
  OID_RSA_SHA256       = '1.2.840.113549.1.1.11' # sha256WithRSAEncryption
  OID_ECDSA_SHA256     = '1.2.840.10045.4.3.2'   # ecdsa-with-SHA256

  PRINTABLE_STRING_RE = %r{\A[A-Za-z0-9 '()+,\-./:=?]*\z}

  class << self
    # --- DER primitives -------------------------------------------------------

    # DER length: short form below 0x80, long form (0x80 | byte count) above.
    def encode_length(len)
      raise ArgumentError, 'Length must be a non-negative integer' unless len.is_a?(Integer) && len >= 0
      return [len].pack('C') if len < 0x80

      bytes = int_to_bytes(len)
      ([0x80 | bytes.bytesize] + bytes.bytes).pack('C*')
    end

    # DER INTEGER. Accepts a non-negative Integer, or raw big-endian bytes (a
    # signature half): leading zeros are stripped and a 0x00 sign byte is
    # prepended when the high bit is set, per DER minimal-encoding rules.
    def encode_integer(value)
      bytes =
        if value.is_a?(Integer)
          raise ArgumentError, 'encode_integer supports non-negative integers only' if value.negative?

          value.zero? ? "\x00".b : int_to_bytes(value)
        else
          strip_leading_zeros(value.to_s.b)
        end
      bytes = "\x00".b + bytes if bytes.getbyte(0) > 0x7f
      tlv(0x02, bytes)
    end

    # DER OBJECT IDENTIFIER from a dotted string, e.g. '1.2.840.113549.1.1.11'.
    def encode_oid(oid)
      parts = oid.to_s.split('.')
      if parts.length < 2 || parts.any? { |p| !p.match?(/\A\d+\z/) }
        raise ArgumentError, "Invalid OID: #{oid}"
      end

      nums = parts.map(&:to_i)
      content = [(40 * nums[0]) + nums[1]]
      nums.drop(2).each { |arc| content.concat(base128(arc)) }
      tlv(0x06, content.pack('C*'))
    end

    def encode_utf8_string(str)
      tlv(0x0c, str.to_s.encode(Encoding::UTF_8).b)
    end

    # PrintableString — the required type for countryName in a Name.
    def encode_printable_string(str)
      raise ArgumentError, "Not a PrintableString: #{str}" unless str.to_s.match?(PRINTABLE_STRING_RE)

      tlv(0x13, str.to_s.b)
    end

    def encode_bit_string(data)
      tlv(0x03, "\x00".b + data.to_s.b) # 0 unused bits in the last octet
    end

    def encode_octet_string(data)
      tlv(0x04, data.to_s.b)
    end

    def encode_null
      "\x05\x00".b
    end

    def encode_sequence(*parts)
      tlv(0x30, parts.join.b)
    end

    def encode_set(*parts)
      tlv(0x31, parts.join.b)
    end

    # --- validation -----------------------------------------------------------

    # Raises on invalid input: CN required, country (when present) a 2-letter
    # ISO 3166-1 code, key algorithm one of the supported values.
    def validate_csr_options(options)
      raise ArgumentError, 'Common Name (CN) is required' if blank?(options.common_name)

      country = options.country.to_s.strip
      unless country.empty? || country.match?(/\A[A-Za-z]{2}\z/)
        raise ArgumentError, 'Country must be a 2-letter ISO 3166-1 code (e.g. US, DE)'
      end

      return if KEY_ALGORITHMS.include?(options.key_algorithm)

      raise ArgumentError, "Unsupported key algorithm: #{options.key_algorithm}"
    end

    # --- subject alternative names --------------------------------------------

    # Classify a SAN entry: IPv4/IPv6 -> :ip, http(s):// -> :uri, contains an
    # '@' -> :email, everything else -> :dns.
    def classify_san_type(entry)
      value = entry.to_s.strip.downcase
      return :ip if ip_to_bytes(value)
      return :uri if value.match?(%r{\Ahttps?://})
      return :email if value.include?('@')

      :dns
    end

    # --- PEM ------------------------------------------------------------------

    # PEM-wrap DER bytes: 64-character base64 lines between BEGIN/END markers.
    def pem_encode(der, label)
      lines = Base64.strict_encode64(der.to_s.b).scan(/.{1,64}/)
      "-----BEGIN #{label}-----\n#{lines.join("\n")}\n-----END #{label}-----\n"
    end

    # --- key generation + CSR assembly ----------------------------------------

    # Generate a key pair and build a signed PKCS#10 CSR (PEM) plus the matching
    # PKCS#8 private key (PEM).
    def generate_csr(options)
      validate_csr_options(options)

      key = generate_key_pair(options.key_algorithm)
      cri = certification_request_info(options, key.public_to_der)

      csr_der = encode_sequence(cri, signature_algorithm(options.key_algorithm),
                                encode_bit_string(key.sign(OpenSSL::Digest::SHA256.new, cri)))

      Result.new(
        csr: pem_encode(csr_der, 'CERTIFICATE REQUEST'),
        private_key: pem_encode(key.private_to_der, 'PRIVATE KEY')
      )
    end

    private

    # --- DER helpers ----------------------------------------------------------

    # Wrap content bytes in a tag + DER length header.
    def tlv(tag, content)
      bytes = content.to_s.b
      [tag].pack('C') + encode_length(bytes.bytesize) + bytes
    end

    # Minimal big-endian byte string for a positive Integer.
    def int_to_bytes(value)
      bytes = []
      n = value
      while n.positive?
        bytes.unshift(n & 0xff)
        n >>= 8
      end
      bytes.pack('C*')
    end

    def strip_leading_zeros(bytes)
      i = 0
      i += 1 while i < bytes.bytesize - 1 && bytes.getbyte(i).zero?
      bytes.byteslice(i..)
    end

    # One OID arc as base-128 septets, continuation bit set on all but the last.
    def base128(arc)
      stack = []
      value = arc
      loop do
        stack.unshift(value & 0x7f)
        value >>= 7
        break if value.zero?
      end
      stack[0..-2].map { |b| b | 0x80 } + [stack[-1]]
    end

    def blank?(value)
      value.to_s.strip.empty?
    end

    # --- subject / attributes -------------------------------------------------

    # AttributeTypeAndValue (SEQUENCE of OID + value) wrapped in its RDN SET.
    def rdn(oid, value)
      encode_set(encode_sequence(encode_oid(oid), value))
    end

    # RDNSequence in the conventional C, ST, L, O, CN, email order.
    def subject(options)
      rdns = []
      country = options.country.to_s.strip.upcase
      rdns << rdn(OID_COUNTRY, encode_printable_string(country)) unless country.empty?
      rdns << rdn(OID_STATE, encode_utf8_string(options.state.to_s.strip)) unless blank?(options.state)
      rdns << rdn(OID_LOCALITY, encode_utf8_string(options.locality.to_s.strip)) unless blank?(options.locality)
      unless blank?(options.organization)
        rdns << rdn(OID_ORGANIZATION, encode_utf8_string(options.organization.to_s.strip))
      end
      rdns << rdn(OID_COMMON_NAME, encode_utf8_string(options.common_name.to_s.strip))
      rdns << rdn(OID_EMAIL, encode_utf8_string(options.email.to_s.strip)) unless blank?(options.email)
      encode_sequence(*rdns)
    end

    # [0] IMPLICIT SET OF Attribute — an extensionRequest carrying the
    # subjectAltName extension. Empty (the field is omitted) when no SANs were
    # requested, matching the TS reference.
    def attributes(options)
      sans = Array(options.subject_alt_names).map { |s| s.to_s.strip }.reject(&:empty?)
      return ''.b if sans.empty?

      extension = encode_sequence(
        encode_oid(OID_SUBJECT_ALT_NAME),
        encode_octet_string(encode_sequence(*sans.map { |san| encode_general_name(san) }))
      )
      tlv(0xa0, encode_sequence(encode_oid(OID_EXT_REQUEST), encode_set(encode_sequence(extension))))
    end

    # One GeneralName: context-specific implicit tags per RFC 5280.
    def encode_general_name(entry)
      value = entry.to_s.strip
      case classify_san_type(value)
      when :ip    then tlv(0x87, ip_to_bytes(value.downcase))
      when :uri   then tlv(0x86, value.b)
      when :email then tlv(0x81, value.b)
      else tlv(0x82, value.b) # dNSName
      end
    end

    # Packed 4- or 16-byte address, or nil when the entry is not an IP literal.
    # A prefix length ('10.0.0.0/8') is not a SAN entry and is rejected.
    def ip_to_bytes(value)
      return nil if value.include?('/')

      IPAddr.new(value).hton
    rescue IPAddr::Error
      nil
    end

    # --- keys + signature -----------------------------------------------------

    def generate_key_pair(algorithm)
      return OpenSSL::PKey::EC.generate('prime256v1') if algorithm == 'ECDSA-P256'

      OpenSSL::PKey::RSA.generate(algorithm == 'RSA-4096' ? 4096 : 2048)
    end

    # CertificationRequestInfo: version 0, subject, SPKI, [0] attributes.
    def certification_request_info(options, spki)
      encode_sequence(encode_integer(0), subject(options), spki, attributes(options))
    end

    # RSA carries an explicit NULL parameter; ECDSA carries none.
    def signature_algorithm(algorithm)
      return encode_sequence(encode_oid(OID_ECDSA_SHA256)) if algorithm == 'ECDSA-P256'

      encode_sequence(encode_oid(OID_RSA_SHA256), encode_null)
    end
  end
end