Add ca-subpath feature and arguments

certipy/commands/parsers/relay.py

@@ -65,6 +65,18 @@ def add_subparser(subparsers: argparse._SubParsersAction) -> Tuple[str, Callable
         ),
     )
 
+    # CA subpath (optional)
+    subparser.add_argument(
+        "-ca-subpath",
+        action="store",
+        metavar="path to ask certificate after /certsrv/",
+        required=False,
+        help=(
+            "Path to ask a certificate after /certsrv/"
+            "Example: /en-us/ for ESC8 (become http://ca.corp.local/certsrv/en-us/certfnsh.asp)"
+        ),
+    )
+
     # Certificate request parameters
     cert_group = subparser.add_argument_group("certificate request options")
     cert_group.add_argument(

certipy/commands/parsers/req.py

@@ -61,6 +61,18 @@ def add_subparser(subparsers: argparse._SubParsersAction) -> Tuple[str, Callable
         help="Name of the Certificate Authority to request certificates from. Required for RPC and DCOM methods",
     )
 
+    # CA subpath (optional)
+    subparser.add_argument(
+        "-ca-subpath",
+        action="store",
+        metavar="path to ask certificate after /certsrv/",
+        required=False,
+        help=(
+            "Path to ask a certificate after /certsrv/"
+            "Example: /en-us/ (become http://ca.corp.local/certsrv/en-us/certfnsh.asp)"
+        ),
+    )
+
     # Certificate request parameters
     cert_group = subparser.add_argument_group("certificate request options")
     cert_group.add_argument(

certipy/commands/relay.py

@@ -400,7 +400,7 @@ class ADCSHTTPAttackClient(ProtocolAttack):
     This class handles certificate operations via the Web Enrollment interface.
     """
 
-    def __init__(self, adcs_relay: "Relay", *args, **kwargs):  # type: ignore
+    def __init__(self, adcs_relay: "Relay", ca_subpath: "", *args, **kwargs):  # type: ignore
         """
         Initialize the HTTP attack client.
 
@@ -411,6 +411,7 @@ def __init__(self, adcs_relay: "Relay", *args, **kwargs):  # type: ignore
         """
         super().__init__(*args, **kwargs)
         self.adcs_relay = adcs_relay
+        self.ca_subpath = ca_subpath
 
     def run(self) -> None:  # type: ignore
         """
@@ -459,7 +460,9 @@ def _enumerate_templates(self) -> None:
         """
 
         # Request the certificate request page
-        res = self.client.get("/certsrv/certrqxt.asp")
+        tmp_path = f"{self.ca_subpath}/" if self.ca_subpath else ""
+        res = self.client.get(f"/certsrv/{tmp_path}certrqxt.asp")
+
         content = res.text
 
         # Parse the HTML to extract templates
@@ -501,6 +504,7 @@ def _retrieve_certificate(self, request_id: int) -> None:
         result = web_retrieve(
             self.client,
             request_id,
+            self.ca_subpath or "",
         )
 
         if result is not None:
@@ -565,6 +569,7 @@ def _request_certificate(self) -> None:
             csr,
             attributes,
             template,
+            self.ca_subpath or "",
             key,
             self.adcs_relay.out,
         )
@@ -790,6 +795,7 @@ def __init__(
         self,
         target: str,
         ca: Optional[str] = None,
+        ca_subpath: Optional[str] = None,
         template: Optional[str] = None,
         upn: Optional[str] = None,
         dns: Optional[str] = None,
@@ -816,6 +822,7 @@ def __init__(
         Args:
             target: Target AD CS server (http://server/certsrv/ or rpc://server)
             ca: Certificate Authority name (required for RPC)
+            ca-subpath: Path to ask certificate after /certsrv/
             template: Certificate template to request
             upn: Alternative UPN (User Principal Name)
             dns: Alternative DNS name
@@ -839,6 +846,7 @@ def __init__(
         self.target = target
         self.base_url = target  # Used only for HTTP(S) targets
         self.ca = ca
+        self.ca_subpath = ca_subpath
         self.template = template
         self.alt_upn = upn
         self.alt_dns = dns
@@ -870,6 +878,10 @@ def __init__(
         self.attacked_targets = []
         self.attack_lock = Lock()
 
+        # Parse the CA subpath
+        if self.ca_subpath is not None:
+            self.ca_subpath = self.ca_subpath.strip("/")
+
         # Configure target based on URL or RPC string
         if self.target.startswith("rpc://"):
             if ca is None:
@@ -883,14 +895,15 @@ def __init__(
                 "https://"
             ):
                 self.target = f"http://{self.target}"
-            if not self.target.endswith("/certsrv/certfnsh.asp"):
+            expected_path = f"/certsrv/{self.ca_subpath + '/' if self.ca_subpath else ''}certfnsh.asp"
+            if not self.target.endswith(expected_path):
                 if not self.target.endswith("/"):
                     self.target += "/"
 
                 if self.enum_templates:
-                    self.target += "certsrv/certrqxt.asp"
+                    self.target += f"certsrv/{(self.ca_subpath + '/') if self.ca_subpath else ''}certrqxt.asp"
                 else:
-                    self.target += "certsrv/certfnsh.asp"
+                    self.target += f"certsrv/{(self.ca_subpath + '/') if self.ca_subpath else ''}certfnsh.asp"
             logging.info(f"Targeting {self.target} (ESC8)")
 
             url = httpx.URL(self.target)

certipy/lib/req.py

@@ -510,6 +510,7 @@ def web_request(
     csr: Union[str, bytes, x509.CertificateSigningRequest],
     attributes_list: List[str],
     template: str,
+    ca_subpath: str,
     key: PrivateKeyTypes,
     out: Optional[str] = None,
 ) -> Optional[x509.Certificate]:
@@ -555,7 +556,10 @@ def web_request(
 
     # Send certificate request
     try:
-        res = session.post("/certsrv/certfnsh.asp", data=params)
+        res = session.post(
+            f"/certsrv/{(ca_subpath + '/') if ca_subpath else ''}certfnsh.asp",
+            data=params,
+        )
         content = res.text
 
         # Handle HTTP errors
@@ -569,7 +573,7 @@ def web_request(
         if request_id_matches:
             request_id = int(request_id_matches[0])
             logging.info(f"Certificate issued with request ID {request_id}")
-            return web_retrieve(session, request_id)
+            return web_retrieve(session, request_id, ca_subpath or "")
 
         # Handle various error conditions
         if "template that is not supported" in content:
@@ -584,7 +588,7 @@ def web_request(
                 if "Certificate Pending" in content:
                     logging.warning("Certificate request is pending approval")
                 elif '"Denied by Policy Module"' in content:
-                    _handle_policy_denial(session, request_id)
+                    _handle_policy_denial(session, request_id, ca_subpath or "")
                 else:
                     _handle_other_errors(content)
             else:
@@ -607,6 +611,7 @@ def web_request(
 def web_retrieve(
     session: httpx.Client,
     request_id: int,
+    ca_subpath: str,
 ) -> Optional[x509.Certificate]:
     """
     Retrieve a certificate via the web enrollment interface.
@@ -622,8 +627,10 @@ def web_retrieve(
 
     try:
         # Request the certificate
-        res = session.get("/certsrv/certnew.cer", params={"ReqID": request_id})
-
+        res = session.get(
+            f"/certsrv/{(ca_subpath + '/') if ca_subpath else ''}certnew.cer",
+            params={"ReqID": request_id},
+        )
         if res.status_code != 200:
             logging.error(f"Error retrieving certificate (HTTP {res.status_code})")
             _log_response_if_verbose(res.text)
@@ -693,7 +700,9 @@ def _determine_output_filename(
     return username.rstrip("$").lower()
 
 
-def _handle_policy_denial(session: httpx.Client, request_id: int) -> None:
+def _handle_policy_denial(
+    session: httpx.Client, request_id: int, ca_subpath: str
+) -> None:
     """
     Handle certificate request denied by policy.
 
@@ -702,8 +711,10 @@ def _handle_policy_denial(session: httpx.Client, request_id: int) -> None:
         request_id: The certificate request ID
     """
     try:
-        res = session.get("/certsrv/certnew.cer", params={"ReqID": request_id})
-
+        res = session.get(
+            f"/certsrv/{(ca_subpath + '/') if ca_subpath else ''}certnew.cer",
+            params={"ReqID": request_id},
+        )
         error_codes = re.findall(
             r"(0x[a-zA-Z0-9]+) \([-]?[0-9]+ ", res.text, flags=re.MULTILINE
         )
@@ -1149,10 +1160,9 @@ def _try_connection(self, session: httpx.Client) -> bool:
         host_value = self.target.remote_name or self.target.target_ip
         if host_value:
             headers["Host"] = host_value
-
         try:
             res = session.get(
-                "/certsrv/",
+                f"/certsrv/{(self.parent.ca_subpath + '/') if self.parent.ca_subpath else ''}",
                 headers=headers,
                 timeout=self.target.timeout,
                 follow_redirects=False,
@@ -1191,10 +1201,7 @@ def retrieve(self, request_id: int) -> Optional[x509.Certificate]:
         if self.session is None:
             raise Exception("Failed to get HTTP session")
 
-        return web_retrieve(
-            self.session,
-            request_id,
-        )
+        return web_retrieve(self.session, request_id, self.parent.ca_subpath or "")
 
     def request(
         self, csr: bytes, attributes_list: List[str]
@@ -1222,6 +1229,7 @@ def request(
             csr,
             attributes_list,
             self.parent.template,
+            self.parent.ca_subpath or "",
             self.parent.key,
             self.parent.out,
         )
@@ -1244,6 +1252,7 @@ def __init__(
         self,
         target: Target,
         ca: Optional[str] = None,
+        ca_subpath: Optional[str] = "",
         template: str = "User",
         upn: Optional[str] = None,
         dns: Optional[str] = None,
@@ -1301,6 +1310,7 @@ def __init__(
         # Core parameters
         self.target = target
         self.ca = ca
+        self.ca_subpath = ca_subpath
         self.template = template
         self.alt_upn = upn
         self.alt_dns = dns
@@ -1317,6 +1327,10 @@ def __init__(
         self.out = out
         self.key = key
 
+        # Parse the CA subpath
+        if self.ca_subpath is not None:
+            self.ca_subpath = self.ca_subpath.strip("/")
+
         # Convert application policy names to OIDs
         self.application_policies = [
             OID_TO_STR_NAME_MAP.get(policy.lower(), policy)

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "certipy-ad"
-version = "5.0.4"
+version = "5.0.5"
 description = "Active Directory Certificate Services enumeration and abuse"
 readme = "README.md"
 authors = [