Make builder/runtime service semantics manifest-driven

src/open_range/builder/builder.py

@@ -14,6 +14,7 @@ import json
 import logging
 import os
 import random
+import re
 from copy import deepcopy
 from pathlib import Path, PurePosixPath
 from typing import Any, Optional
@@ -40,7 +41,10 @@ from open_range.protocols import (
 )
 
 from open_range.builder.prompts import BUILDER_SYSTEM_PROMPT
-from open_range.builder.manifest_graph import compile_manifest_topology
+from open_range.builder.manifest_graph import (
+    compile_manifest_topology,
+    runtime_contract_from_topology,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -569,7 +573,8 @@ def _parse_llm_response(raw_json: str) -> SnapshotSpec:
         elif isinstance(vc, str) and vc.strip():
             ip = v.injection_point
             if ip.startswith("/") and v.host == "web":
-                container_key = f"web:/var/www/portal{ip}"
+                web_doc_root = str(llm_output.topology.get("web_doc_root", "/var/www/portal")).rstrip("/")
+                container_key = f"web:{web_doc_root}{ip}"
                 if container_key not in files:
                     files[container_key] = vc
 
@@ -893,13 +898,13 @@ _DEFAULT_VULN_POOL: list[dict[str, Any]] = [
             },
             {
                 "step": 3,
-                "cmd": "ldapsearch -x -H ldap://ldap -D 'cn=webapp,dc=corp,dc=local' -w 'Svc!Ldap2024' -b 'dc=corp,dc=local' '(uid=*)'",
+                "cmd": "ldapsearch -x -H ldap://{ldap_host} -D '{ldap_bind_dn}' -w '{ldap_bind_pw}' -b '{ldap_search_base_dn}' '(uid=*)'",
                 "expect_stdout": "dn: uid=",
                 "host": "attacker",
             },
             {
                 "step": 4,
-                "cmd": "sshpass -p 'Svc!Ldap2024' ssh svc_webapp@db 'cat /var/flags/flag1.txt'",
+                "cmd": "sshpass -p '{credential_reuse_password}' ssh {credential_reuse_user}@{credential_reuse_host} 'cat /var/flags/flag1.txt'",
                 "expect_stdout": "FLAG{cr3d_r3us3_p1v0t}",
                 "host": "attacker",
             },
@@ -998,7 +1003,7 @@ class TemplateOnlyBuilder:
                 z = h.get("zone", "default")
                 zones.setdefault(z, []).append(h["name"])
 
-        topology = {
+        topology: dict[str, Any] = {
             "tier": int(manifest.get("tier", context.tier) or context.tier),
             "hosts": hosts,
             "zones": zones,
@@ -1012,6 +1017,9 @@ class TemplateOnlyBuilder:
             ),
             "mysql_root_password": "root" if any(v["type"] == "weak_creds" for v in chosen) else "r00tP@ss!",
         }
+        topology = compile_manifest_topology(manifest, topology)
+        runtime_contract = runtime_contract_from_topology(topology, manifest=manifest)
+        topology["runtime_contract"] = runtime_contract
 
         # Build truth graph
         vulns = []
@@ -1020,7 +1028,12 @@ class TemplateOnlyBuilder:
         golden_path: list[GoldenPathStep] = []
         step_offset = 0
 
-        for idx, v in enumerate(chosen):
+        for idx, raw in enumerate(chosen):
+            v = _realize_template_vuln(
+                raw,
+                topology=topology,
+                runtime_contract=runtime_contract,
+            )
             vulns.append(
                 Vulnerability(
                     id=v["id"],
@@ -1038,7 +1051,7 @@ class TemplateOnlyBuilder:
                 {
                     "vuln_id": v["id"],
                     "command": v.get("injection_point", ""),
-                    "description": f"Exploit {v['type']} on {v['host']}",
+                    "description": f"Exploit {v['type']} on {v.get('host', 'target')}",
                 }
             )
             flags.append(
@@ -1046,7 +1059,7 @@ class TemplateOnlyBuilder:
                     id=v.get("flag_id", f"flag{idx+1}"),
                     value=v.get("flag_value", f"FLAG{{test_{idx+1}}}"),
                     path=v.get("flag_path", f"/var/flags/flag{idx+1}.txt"),
-                    host=v.get("host", "web"),
+                    host=v.get("flag_host", v.get("host", runtime_contract["web_host"])),
                 )
             )
             for gs in v.get("golden_path_steps", []):
@@ -1060,6 +1073,7 @@ class TemplateOnlyBuilder:
                         step=step_offset,
                         command=cmd,
                         expect_in_stdout=gs["expect_stdout"],
+                        host=gs.get("host", "attacker"),
                         description=gs.get("description", ""),
                     )
                 )
@@ -1069,7 +1083,7 @@ class TemplateOnlyBuilder:
         evidence_spec = [
             EvidenceItem(
                 type="log_entry",
-                location="web:/var/log/app/access.log",
+                location=f"{runtime_contract['web_host']}:/var/log/app/access.log",
                 pattern="attack pattern from attacker IP",
             ),
             EvidenceItem(
@@ -1125,6 +1139,179 @@ class TemplateOnlyBuilder:
 # ---------------------------------------------------------------------------
 
 
+def _realize_template_vuln(
+    template: dict[str, Any],
+    *,
+    topology: dict[str, Any],
+    runtime_contract: dict[str, str],
+) -> dict[str, Any]:
+    realized = deepcopy(template)
+    template_host = str(template.get("host", "")).strip()
+    service = str(template.get("service", "")).strip().lower()
+    resolved_host = _resolve_vuln_host(
+        template_host,
+        service=service,
+        topology=topology,
+        runtime_contract=runtime_contract,
+    )
+    realized["host"] = resolved_host
+
+    vuln_type = str(template.get("type", "")).strip()
+    if vuln_type == "credential_reuse":
+        realized["flag_host"] = runtime_contract.get(
+            "credential_reuse_host",
+            runtime_contract.get("db_host", resolved_host),
+        )
+    else:
+        realized["flag_host"] = resolved_host
+
+    for field in (
+        "injection_point",
+        "vulnerable_code",
+        "root_cause",
+        "blast_radius",
+        "remediation",
+    ):
+        value = realized.get(field)
+        if isinstance(value, str):
+            realized[field] = _rewrite_template_runtime_text(value, runtime_contract)
+
+    raw_steps = template.get("golden_path_steps", [])
+    realized_steps: list[dict[str, Any]] = []
+    if isinstance(raw_steps, list):
+        for raw_step in raw_steps:
+            if not isinstance(raw_step, dict):
+                continue
+            step = deepcopy(raw_step)
+            cmd = str(step.get("cmd", ""))
+            expect = str(step.get("expect_stdout", ""))
+            step["cmd"] = _rewrite_template_runtime_text(cmd, runtime_contract)
+            step["expect_stdout"] = _rewrite_template_runtime_text(expect, runtime_contract)
+            realized_steps.append(step)
+    realized["golden_path_steps"] = realized_steps
+    return realized
+
+
+def _resolve_vuln_host(
+    template_host: str,
+    *,
+    service: str,
+    topology: dict[str, Any],
+    runtime_contract: dict[str, str],
+) -> str:
+    hosts = _host_names(topology.get("hosts", []))
+    alias_map = {
+        "web": runtime_contract.get("web_host", "web"),
+        "db": runtime_contract.get("db_host", "db"),
+        "ldap": runtime_contract.get("ldap_host", "ldap"),
+    }
+    if template_host:
+        if template_host in hosts:
+            return template_host
+        if template_host in alias_map and alias_map[template_host]:
+            return alias_map[template_host]
+
+    if any(marker in service for marker in ("mysql", "mariadb", "postgres")):
+        candidate = runtime_contract.get("db_host", "db")
+        if not hosts or candidate in hosts:
+            return candidate
+    if any(marker in service for marker in ("ldap", "openldap")):
+        candidate = runtime_contract.get("ldap_host", "ldap")
+        if not hosts or candidate in hosts:
+            return candidate
+    if any(marker in service for marker in ("nginx", "apache", "http", "php")):
+        candidate = runtime_contract.get("web_host", "web")
+        if not hosts or candidate in hosts:
+            return candidate
+
+    if template_host:
+        return template_host
+    if hosts:
+        return hosts[0]
+    return runtime_contract.get("web_host", "web")
+
+
+def _host_names(raw_hosts: object) -> list[str]:
+    if not isinstance(raw_hosts, list):
+        return []
+    hosts: list[str] = []
+    for raw in raw_hosts:
+        if isinstance(raw, dict):
+            host = str(raw.get("name", "")).strip()
+        else:
+            host = str(raw).strip()
+        if host and host not in hosts:
+            hosts.append(host)
+    return hosts
+
+
+def _rewrite_template_runtime_text(text: str, runtime_contract: dict[str, str]) -> str:
+    if not text:
+        return text
+
+    web_host = runtime_contract.get("web_host", "web")
+    db_host = runtime_contract.get("db_host", "db")
+    ldap_host = runtime_contract.get("ldap_host", "ldap")
+    web_doc_root = runtime_contract.get("web_doc_root", "/var/www/portal")
+    web_config_path = runtime_contract.get("web_config_path", "/var/www/config.php")
+    db_name = runtime_contract.get("db_name", "referral_db")
+    db_user = runtime_contract.get("db_user", "svc_db")
+    db_password = runtime_contract.get("db_password", "SvcDb!401")
+    ldap_bind_dn = runtime_contract.get("ldap_bind_dn", f"cn={db_user},dc=corp,dc=local")
+    ldap_bind_pw = runtime_contract.get("ldap_bind_pw", db_password)
+    reuse_user = runtime_contract.get("credential_reuse_user", db_user)
+    reuse_host = runtime_contract.get("credential_reuse_host", db_host)
+    reuse_password = runtime_contract.get("credential_reuse_password", ldap_bind_pw)
+
+    updated = text
+    placeholders = {
+        "{web_host}": web_host,
+        "{db_host}": db_host,
+        "{ldap_host}": ldap_host,
+        "{web_doc_root}": web_doc_root,
+        "{web_config_path}": web_config_path.lstrip("/"),
+        "{db_name}": db_name,
+        "{db_user}": db_user,
+        "{db_password}": db_password,
+        "{ldap_bind_dn}": ldap_bind_dn,
+        "{ldap_bind_pw}": ldap_bind_pw,
+        "{ldap_search_base_dn}": runtime_contract.get("ldap_search_base_dn", "dc=corp,dc=local"),
+        "{credential_reuse_user}": reuse_user,
+        "{credential_reuse_host}": reuse_host,
+        "{credential_reuse_password}": reuse_password,
+    }
+    for placeholder, value in placeholders.items():
+        updated = updated.replace(placeholder, value)
+
+    replacements: list[tuple[str, str]] = [
+        ("http://web/", f"http://{web_host}/"),
+        ("http://web", f"http://{web_host}"),
+        ("ldap://ldap", f"ldap://{ldap_host}"),
+        ("svc_webapp@db", f"{reuse_user}@{reuse_host}"),
+        ("@db ", f"@{db_host} "),
+        ("@db'", f"@{db_host}'"),
+        ('@db"', f'@{db_host}"'),
+        (" -h db ", f" -h {db_host} "),
+        (" -h db", f" -h {db_host}"),
+        ("/var/www/portal", web_doc_root),
+        ("/var/www/config.php", web_config_path),
+        ("referral_db", db_name),
+        ("app_user", db_user),
+        ("AppUs3r!2024", db_password),
+        ("Svc!Ldap2024", ldap_bind_pw),
+    ]
+    for old, new in replacements:
+        updated = updated.replace(old, new)
+
+    updated = updated.replace("cn=webapp,dc=corp,dc=local", ldap_bind_dn)
+    updated = re.sub(
+        r"cn=webapp,dc=[A-Za-z0-9_-]+(?:,dc=[A-Za-z0-9_-]+)*",
+        ldap_bind_dn,
+        updated,
+    )
+    return updated
+
+
 def _manifest_topology_users(
     manifest: dict[str, Any],
     *,
@@ -1194,6 +1381,7 @@ def render_template_payloads(
     manifest: dict[str, Any] | None = None,
 ) -> dict[str, str]:
     topology = snapshot.topology if isinstance(snapshot.topology, dict) else {}
+    runtime_contract = runtime_contract_from_topology(topology, manifest=manifest)
     flags = snapshot.flags
     evidence_spec = snapshot.evidence_spec
     vuln_types = {v.type for v in snapshot.truth_graph.vulns}
@@ -1204,31 +1392,46 @@ def render_template_payloads(
     )
     company_name = str(topology.get("org_name") or company.get("name") or "OpenRange")
     domain = str(topology.get("domain") or company.get("domain") or "corp.local")
+    web_host = runtime_contract["web_host"]
+    db_host = runtime_contract["db_host"]
+    web_doc_root = runtime_contract["web_doc_root"]
+    web_config_path = runtime_contract["web_config_path"]
+    db_name = runtime_contract["db_name"]
 
     files: dict[str, str] = {
-        "web:/var/www/portal/index.php": _default_index_php(company_name),
-        "web:/var/www/portal/login.php": _default_login_php(),
-        "web:/var/www/config.php": _default_config_php(domain=domain),
+        f"{web_host}:{_join_posix(web_doc_root, 'index.php')}": _default_index_php(company_name),
+        f"{web_host}:{_join_posix(web_doc_root, 'login.php')}": _default_login_php(),
+        f"{web_host}:{web_config_path}": _default_config_php(
+            domain=domain,
+            db_host=runtime_contract["db_host"],
+            db_name=runtime_contract["db_name"],
+            db_user=runtime_contract["db_user"],
+            db_pass=runtime_contract["db_password"],
+            ldap_bind_dn=runtime_contract["ldap_bind_dn"],
+            ldap_bind_pw=runtime_contract["ldap_bind_pw"],
+        ),
     }
 
     if "sqli" in vuln_types:
-        files["web:/var/www/portal/search.php"] = _search_php(
+        files[f"{web_host}:{_join_posix(web_doc_root, 'search.php')}"] = _search_php(
             _flag_value_for_type(snapshot, "sqli")
         )
 
     if "path_traversal" in vuln_types:
-        files["web:/var/www/portal/download.php"] = _download_php(
+        files[f"{web_host}:{_join_posix(web_doc_root, 'download.php')}"] = _download_php(
             path_flag=_flag_value_for_type(snapshot, "path_traversal"),
             flag_names=_flag_names_for_type(snapshot, "path_traversal"),
+            config_path=web_config_path,
         )
     elif "credential_reuse" in vuln_types:
-        files["web:/var/www/portal/download.php"] = _download_php(
+        files[f"{web_host}:{_join_posix(web_doc_root, 'download.php')}"] = _download_php(
             path_flag="",
             flag_names=[],
+            config_path=web_config_path,
         )
 
     if "idor" in vuln_types:
-        files["web:/var/www/portal/api/index.php"] = _idor_api_php(
+        files[f"{web_host}:{_join_posix(web_doc_root, 'api/index.php')}"] = _idor_api_php(
             _flag_value_for_type(snapshot, "idor"),
         )
 
@@ -1249,7 +1452,7 @@ def render_template_payloads(
                         "CREATE USER IF NOT EXISTS 'leaked_user'@'%' "
                         "IDENTIFIED BY 'leaked_pass';\n"
                         "GRANT SELECT ON flags.* TO 'leaked_user'@'%';\n"
-                        "GRANT SELECT ON referral_db.* TO 'leaked_user'@'%';\n"
+                        f"GRANT SELECT ON {_sql_ident(db_name)}.* TO 'leaked_user'@'%';\n"
                         "FLUSH PRIVILEGES;\n"
                     ),
                 )
@@ -1265,7 +1468,7 @@ def render_template_payloads(
         )
 
     if "weak_creds" in vuln_types:
-        files["db:/tmp/openrange-root-password.txt"] = "root\n"
+        files[f"{db_host}:/tmp/openrange-root-password.txt"] = "root\n"
 
     return files
 
@@ -1315,6 +1518,15 @@ def _append_sql(existing: str, fragment: str) -> str:
     return f"{existing.rstrip()}\n{fragment}"
 
 
+def _join_posix(base: str, leaf: str) -> str:
+    return (PurePosixPath(base) / leaf).as_posix()
+
+
+def _sql_ident(value: str) -> str:
+    token = re.sub(r"[^A-Za-z0-9_]", "", value)
+    return token or "referral_db"
+
+
 def _sql_escape(value: str) -> str:
     """Escape a string for use in a SQL single-quoted literal.
 
@@ -1356,15 +1568,26 @@ echo "Login";
 ?>"""
 
 
-def _default_config_php(*, domain: str) -> str:
+def _default_config_php(
+    *,
+    domain: str,
+    db_host: str,
+    db_name: str,
+    db_user: str,
+    db_pass: str,
+    ldap_bind_dn: str,
+    ldap_bind_pw: str,
+) -> str:
     ldap_dn = ",".join(f"dc={part}" for part in domain.split(".") if part) or "dc=corp,dc=local"
+    bind_dn = ldap_bind_dn or f"cn={db_user},{ldap_dn}"
+    bind_pw = ldap_bind_pw or db_pass
     return f"""<?php
-$db_host = "db";
-$db_name = "referral_db";
-$db_user = "app_user";
-$db_pass = "AppUs3r!2024";
-$ldap_bind_dn = "cn=webapp,{ldap_dn}";
-$ldap_bind_pw = "Svc!Ldap2024";
+$db_host = "{db_host}";
+$db_name = "{db_name}";
+$db_user = "{db_user}";
+$db_pass = "{db_pass}";
+$ldap_bind_dn = "{bind_dn}";
+$ldap_bind_pw = "{bind_pw}";
 ?>"""
 
 
@@ -1379,7 +1602,12 @@ if (stripos($q, "union") !== false || stripos($q, "flag") !== false) {{
 ?>"""
 
 
-def _download_php(path_flag: str, flag_names: list[str] | None = None) -> str:
+def _download_php(
+    path_flag: str,
+    flag_names: list[str] | None = None,
+    *,
+    config_path: str,
+) -> str:
     flag = path_flag or "FLAG{placeholder}"
     raw_names = ["flag1.txt"] if flag_names is None else flag_names
     cases = "\n".join(
@@ -1393,7 +1621,7 @@ $file = $_GET["file"] ?? "";
 if ($file === "report.pdf") {{
     echo "PDF";
 }} elseif (strpos($file, "config.php") !== false) {{
-    readfile("/var/www/config.php");
+    readfile("{config_path}");
 }} elseif (strpos($file, "/etc/passwd") !== false) {{
     echo "root:x:0:0:root:/root:/bin/bash";
 }} {cases} else {{

src/open_range/builder/manifest_graph.py

@@ -10,6 +10,8 @@ accounts in rendered services.
 from __future__ import annotations
 
 from copy import deepcopy
+from pathlib import PurePosixPath
+import re
 from typing import Any
 
 
@@ -148,9 +150,352 @@ def compile_manifest_topology(
         if trust_only
         else [],
     }
+    runtime_contract = runtime_contract_from_topology(compiled, manifest=manifest)
+    compiled["runtime_contract"] = runtime_contract
+    compiled.setdefault("web_host", runtime_contract["web_host"])
+    compiled.setdefault("db_host", runtime_contract["db_host"])
+    compiled.setdefault("ldap_host", runtime_contract["ldap_host"])
+    compiled.setdefault("web_doc_root", runtime_contract["web_doc_root"])
+    compiled.setdefault("web_config_path", runtime_contract["web_config_path"])
+    compiled.setdefault("db_name", runtime_contract["db_name"])
+    compiled.setdefault("db_user", runtime_contract["db_user"])
+    compiled.setdefault("db_pass", runtime_contract["db_password"])
+    compiled.setdefault("db_password", runtime_contract["db_password"])
+    compiled.setdefault("ldap_bind_dn", runtime_contract["ldap_bind_dn"])
+    compiled.setdefault("ldap_bind_pw", runtime_contract["ldap_bind_pw"])
+    compiled.setdefault("ldap_search_base_dn", runtime_contract["ldap_search_base_dn"])
+    compiled.setdefault("credential_reuse_user", runtime_contract["credential_reuse_user"])
+    compiled.setdefault("credential_reuse_host", runtime_contract["credential_reuse_host"])
+    compiled.setdefault(
+        "credential_reuse_password",
+        runtime_contract["credential_reuse_password"],
+    )
+
+    service_accounts = compiled.get("service_accounts")
+    if not isinstance(service_accounts, dict):
+        service_accounts = {}
+    webapp = service_accounts.get("webapp")
+    if not isinstance(webapp, dict):
+        webapp = {}
+    webapp.setdefault("username", runtime_contract["db_user"])
+    webapp.setdefault("password", runtime_contract["db_password"])
+    webapp.setdefault("ldap_bind_dn", runtime_contract["ldap_bind_dn"])
+    webapp.setdefault("ldap_bind_pw", runtime_contract["ldap_bind_pw"])
+    service_accounts["webapp"] = webapp
+    compiled["service_accounts"] = service_accounts
     return compiled
 
 
+def runtime_contract_from_topology(
+    topology: dict[str, Any] | None,
+    *,
+    manifest: dict[str, Any] | None = None,
+) -> dict[str, str]:
+    """Derive runtime service/account semantics from compiled topology state."""
+    source = topology if isinstance(topology, dict) else {}
+    runtime = deepcopy(source.get("runtime_contract", {}))
+    if not isinstance(runtime, dict):
+        runtime = {}
+
+    domain = _coerce_text(
+        runtime.get("domain"),
+        source.get("domain"),
+        _manifest_company_domain(manifest),
+        default="corp.local",
+    )
+    host_catalog = source.get("host_catalog", {})
+    if not isinstance(host_catalog, dict):
+        host_catalog = {}
+    host_details = source.get("host_details", {})
+    if not isinstance(host_details, dict):
+        host_details = {}
+    hosts = _normalized_hosts(source.get("hosts", []))
+
+    web_host = _select_core_host(
+        explicit=_coerce_text(runtime.get("web_host"), source.get("web_host")),
+        hosts=hosts,
+        host_maps=[host_catalog, host_details],
+        preferred_names=("web", "portal", "frontend"),
+        service_markers=("nginx", "apache", "http", "php", "gunicorn", "uvicorn"),
+        fallback="web",
+    )
+    db_host = _select_core_host(
+        explicit=_coerce_text(runtime.get("db_host"), source.get("db_host")),
+        hosts=hosts,
+        host_maps=[host_catalog, host_details],
+        preferred_names=("db", "database", "mysql"),
+        service_markers=("mysql", "mariadb", "postgres", "postgresql", "database"),
+        fallback="db",
+    )
+    ldap_host = _select_core_host(
+        explicit=_coerce_text(runtime.get("ldap_host"), source.get("ldap_host")),
+        hosts=hosts,
+        host_maps=[host_catalog, host_details],
+        preferred_names=("ldap", "directory", "idp"),
+        service_markers=("ldap", "openldap"),
+        fallback="ldap",
+    )
+
+    db_name = _coerce_text(
+        runtime.get("db_name"),
+        source.get("db_name"),
+        _infer_manifest_db_name(manifest),
+        default="referral_db",
+    )
+
+    service_accounts = source.get("service_accounts", {})
+    if not isinstance(service_accounts, dict):
+        service_accounts = {}
+    webapp_account = service_accounts.get("webapp", {})
+    if not isinstance(webapp_account, dict):
+        webapp_account = {}
+
+    db_user = _coerce_text(
+        runtime.get("db_user"),
+        source.get("db_user"),
+        source.get("db_app_user"),
+        webapp_account.get("username"),
+    )
+    db_password = _coerce_text(
+        runtime.get("db_password"),
+        runtime.get("db_pass"),
+        source.get("db_password"),
+        source.get("db_pass"),
+        source.get("db_app_password"),
+        webapp_account.get("password"),
+    )
+
+    users = source.get("users", [])
+    selected_user, selected_password = _pick_db_account(users, db_host)
+    if not db_user:
+        db_user = selected_user
+    if not db_password:
+        db_password = selected_password
+    if not db_user:
+        db_user = f"svc_{_slug_token(db_host or 'db')}"
+    if not db_password:
+        db_password = _predictable_service_password(db_user, domain)
+
+    web_doc_root = _coerce_text(
+        runtime.get("web_doc_root"),
+        source.get("web_doc_root"),
+        default="/var/www/portal",
+    )
+    if not web_doc_root.startswith("/"):
+        web_doc_root = f"/{web_doc_root}"
+    web_doc_parent = PurePosixPath(web_doc_root).parent
+    default_config_path = (web_doc_parent / "config.php").as_posix()
+    if not default_config_path.startswith("/"):
+        default_config_path = "/var/www/config.php"
+    web_config_path = _coerce_text(
+        runtime.get("web_config_path"),
+        source.get("web_config_path"),
+        default=default_config_path,
+    )
+    if not web_config_path.startswith("/"):
+        web_config_path = f"/{web_config_path}"
+
+    ldap_base_dn = _domain_to_ldap_dn(domain)
+    ldap_search_base_dn = _coerce_text(
+        runtime.get("ldap_search_base_dn"),
+        source.get("ldap_search_base_dn"),
+        default=ldap_base_dn,
+    )
+    ldap_bind_dn = _coerce_text(
+        runtime.get("ldap_bind_dn"),
+        source.get("ldap_bind_dn"),
+        webapp_account.get("ldap_bind_dn"),
+        default=f"cn={db_user},{ldap_base_dn}",
+    )
+    ldap_bind_pw = _coerce_text(
+        runtime.get("ldap_bind_pw"),
+        source.get("ldap_bind_pw"),
+        webapp_account.get("ldap_bind_pw"),
+        default=db_password,
+    )
+
+    credential_reuse_user = _coerce_text(
+        runtime.get("credential_reuse_user"),
+        source.get("credential_reuse_user"),
+        default=db_user,
+    )
+    credential_reuse_host = _coerce_text(
+        runtime.get("credential_reuse_host"),
+        source.get("credential_reuse_host"),
+        default=db_host,
+    )
+    credential_reuse_password = _coerce_text(
+        runtime.get("credential_reuse_password"),
+        source.get("credential_reuse_password"),
+        default=ldap_bind_pw,
+    )
+
+    return {
+        "domain": domain,
+        "web_host": web_host,
+        "db_host": db_host,
+        "ldap_host": ldap_host,
+        "web_doc_root": web_doc_root,
+        "web_config_path": web_config_path,
+        "db_name": db_name,
+        "db_user": db_user,
+        "db_password": db_password,
+        "ldap_bind_dn": ldap_bind_dn,
+        "ldap_bind_pw": ldap_bind_pw,
+        "ldap_search_base_dn": ldap_search_base_dn,
+        "credential_reuse_user": credential_reuse_user,
+        "credential_reuse_host": credential_reuse_host,
+        "credential_reuse_password": credential_reuse_password,
+    }
+
+
+def _manifest_company_domain(manifest: dict[str, Any] | None) -> str:
+    if not isinstance(manifest, dict):
+        return ""
+    company = manifest.get("company", {})
+    if not isinstance(company, dict):
+        return ""
+    return str(company.get("domain", "")).strip()
+
+
+def _normalized_hosts(raw_hosts: object) -> list[str]:
+    hosts: list[str] = []
+    if not isinstance(raw_hosts, list):
+        return hosts
+    for raw in raw_hosts:
+        if isinstance(raw, dict):
+            name = str(raw.get("name", "")).strip()
+        else:
+            name = str(raw).strip()
+        if name and name not in hosts:
+            hosts.append(name)
+    return hosts
+
+
+def _select_core_host(
+    *,
+    explicit: str,
+    hosts: list[str],
+    host_maps: list[dict[str, Any]],
+    preferred_names: tuple[str, ...],
+    service_markers: tuple[str, ...],
+    fallback: str,
+) -> str:
+    if explicit and (not hosts or explicit in hosts):
+        return explicit
+    for name in preferred_names:
+        if name in hosts:
+            return name
+    for host in hosts:
+        services = _host_services(host, host_maps)
+        if not services:
+            continue
+        if any(
+            marker in service
+            for service in services
+            for marker in service_markers
+        ):
+            return host
+    for host in hosts:
+        lowered = host.lower()
+        if any(name in lowered for name in preferred_names):
+            return host
+    if hosts:
+        return hosts[0]
+    return fallback
+
+
+def _host_services(host: str, host_maps: list[dict[str, Any]]) -> list[str]:
+    services: list[str] = []
+    for host_map in host_maps:
+        detail = host_map.get(host, {})
+        if not isinstance(detail, dict):
+            continue
+        raw_services = detail.get("services", [])
+        if not isinstance(raw_services, list):
+            continue
+        for raw_service in raw_services:
+            service = str(raw_service).strip().lower()
+            if service and service not in services:
+                services.append(service)
+    return services
+
+
+def _pick_db_account(raw_users: object, db_host: str) -> tuple[str, str]:
+    if not isinstance(raw_users, list):
+        return "", ""
+    for raw in raw_users:
+        if not isinstance(raw, dict):
+            continue
+        username = str(raw.get("username", "")).strip()
+        if not username:
+            continue
+        hosts = raw.get("hosts", [])
+        if not isinstance(hosts, list) or db_host not in hosts:
+            continue
+        password = str(raw.get("password", "")).strip()
+        if not _is_privileged_account(raw):
+            return username, password
+    return "", ""
+
+
+def _is_privileged_account(user: dict[str, Any]) -> bool:
+    groups = user.get("groups", [])
+    if isinstance(groups, list):
+        lowered = {str(group).strip().lower() for group in groups}
+        if {"admin", "admins"} & lowered:
+            return True
+    role = str(user.get("role", "")).lower()
+    return "admin" in role
+
+
+def _infer_manifest_db_name(manifest: dict[str, Any] | None) -> str:
+    if not isinstance(manifest, dict):
+        return ""
+    for raw in manifest.get("data_inventory", []):
+        if not isinstance(raw, dict):
+            continue
+        location = str(raw.get("location", "")).strip()
+        lowered = location.lower()
+        for prefix in ("mysql:", "db:"):
+            if not lowered.startswith(prefix):
+                continue
+            raw_name = location[len(prefix):].split(".", 1)[0].strip()
+            if raw_name:
+                return raw_name
+    return ""
+
+
+def _domain_to_ldap_dn(domain: str) -> str:
+    parts = [part for part in domain.split(".") if part]
+    if not parts:
+        return "dc=corp,dc=local"
+    return ",".join(f"dc={part}" for part in parts)
+
+
+def _predictable_service_password(username: str, domain: str) -> str:
+    token = _slug_token(username).replace("_", "")
+    if not token:
+        token = "service"
+    suffix = 200 + (sum(ord(ch) for ch in f"{username}:{domain}") % 700)
+    return f"{token.capitalize()}!{suffix}"
+
+
+def _slug_token(value: str) -> str:
+    token = re.sub(r"[^a-z0-9]+", "_", value.lower()).strip("_")
+    return token or "service"
+
+
+def _coerce_text(*values: object, default: str = "") -> str:
+    for value in values:
+        if value is None:
+            continue
+        text = str(value).strip()
+        if text:
+            return text
+    return default
+
+
 def _merge_hosts(
     raw_hosts: object,
     host_catalog: dict[str, dict[str, Any]],

src/open_range/builder/renderer.py

@@ -16,6 +16,7 @@ from typing import Any
 
 import jinja2
 
+from open_range.builder.manifest_graph import runtime_contract_from_topology
 from open_range.builder.service_manifest import generate_service_specs
 from open_range.protocols import SnapshotSpec
 
@@ -161,6 +162,7 @@ def _build_context(spec: SnapshotSpec) -> dict[str, Any]:
     hosts_raw = topology.get("hosts", [])
     zones = topology.get("zones", {})
     users = topology.get("users", [])
+    runtime_contract = runtime_contract_from_topology(topology)
 
     # Build host objects with name, zone, networks, depends_on
     hosts = _build_hosts(hosts_raw, zones)
@@ -208,8 +210,8 @@ def _build_context(spec: SnapshotSpec) -> dict[str, Any]:
         has_download,
     )
 
-    db_user = _find_db_user(users)
-    db_pass = _find_db_pass(users)
+    db_user = runtime_contract["db_user"]
+    db_pass = runtime_contract["db_password"]
 
     context: dict[str, Any] = {
         # docker-compose.yml.j2
@@ -217,26 +219,31 @@ def _build_context(spec: SnapshotSpec) -> dict[str, Any]:
         "networks": networks,
         "hosts": hosts,
         "host_names": host_names,
-        "db_host": "db",
+        "db_host": runtime_contract["db_host"],
         "db_user": db_user,
         "db_pass": db_pass,
-        "db_name": topology.get("db_name", "app_db"),
+        "db_name": runtime_contract["db_name"],
         # db_password duplicates db_pass: Dockerfile.db.j2 uses db_pass,
         # docker-compose.yml.j2 uses db_password.  Keep both for compat.
         "db_password": db_pass,
         "mysql_root_password": topology.get("mysql_root_password", _find_mysql_root_pass(users)),
-        "domain": topology.get("domain", "corp.local"),
+        "domain": runtime_contract["domain"],
         "org_name": topology.get("org_name", "Corp"),
         "ldap_admin_pass": topology.get("ldap_admin_pass", "LdapAdm1n!"),
         "smb_shares": _find_smb_shares(spec),
         "smb_user": _find_smb_user(users),
         "smb_password": _find_smb_pass(users),
+        "web_doc_root": runtime_contract["web_doc_root"],
+        "web_config_path": runtime_contract["web_config_path"],
+        "ldap_bind_dn": runtime_contract["ldap_bind_dn"],
+        "ldap_bind_pw": runtime_contract["ldap_bind_pw"],
+        "ldap_search_base_dn": runtime_contract["ldap_search_base_dn"],
         # Dockerfile.web.j2
         "users": users,
         "app_files": app_files,
         "flags": flags,
         # nginx.conf.j2
-        "server_name": topology.get("domain", "web.corp.local"),
+        "server_name": topology.get("domain", f"{runtime_contract['web_host']}.{runtime_contract['domain']}"),
         # iptables.rules.j2
         "firewall_rules": firewall_rules,
         "zone_cidrs": zone_cidrs,

src/open_range/builder/templates/Dockerfile.db.j2

@@ -1,9 +1,9 @@
 FROM mysql:8.0
 
 ENV MYSQL_ROOT_PASSWORD={{ mysql_root_password | default('r00tP@ss!') }}
-ENV MYSQL_DATABASE=referral_db
-ENV MYSQL_USER={{ db_user | default('app_user') }}
-ENV MYSQL_PASSWORD={{ db_pass | default('AppUs3r!2024') }}
+ENV MYSQL_DATABASE={{ db_name | default('referral_db') }}
+ENV MYSQL_USER={{ db_user | default('svc_db') }}
+ENV MYSQL_PASSWORD={{ db_pass | default('SvcDb!401') }}
 
 # Copy initialization SQL
 COPY init.sql /docker-entrypoint-initdb.d/01-init.sql

src/open_range/builder/templates/docker-compose.yml.j2

@@ -125,9 +125,9 @@ services:
     command: --default-authentication-plugin=mysql_native_password
     environment:
       - MYSQL_ROOT_PASSWORD={{ mysql_root_password | default('r00tP@ss!') }}
-      - MYSQL_DATABASE={{ db_name | default('app_db') }}
-      - MYSQL_USER={{ db_user | default('app_user') }}
-      - MYSQL_PASSWORD={{ db_password | default('AppUs3r!2024') }}
+      - MYSQL_DATABASE={{ db_name | default('referral_db') }}
+      - MYSQL_USER={{ db_user | default('svc_db') }}
+      - MYSQL_PASSWORD={{ db_password | default('SvcDb!401') }}
     volumes:
       - db_data:/var/lib/mysql
       - shared_logs:/var/log/mysql

src/open_range/builder/templates/init.sql.j2

@@ -1,9 +1,9 @@
 -- OpenRange database initialization
 -- Generated from SnapshotSpec
 
--- Application database (referral_db)
-CREATE DATABASE IF NOT EXISTS referral_db;
-USE referral_db;
+-- Application database
+CREATE DATABASE IF NOT EXISTS {{ db_name | default('referral_db') }};
+USE {{ db_name | default('referral_db') }};
 
 -- Users table (web app authentication)
 CREATE TABLE IF NOT EXISTS users (
@@ -86,9 +86,9 @@ CREATE TABLE IF NOT EXISTS secrets (
     created_at DATETIME DEFAULT CURRENT_TIMESTAMP
 );
 
--- Create and grant app_user (used by PHP web app)
-CREATE USER IF NOT EXISTS 'app_user'@'%' IDENTIFIED WITH mysql_native_password BY 'AppUs3r!2024';
-GRANT SELECT, INSERT, UPDATE ON referral_db.* TO 'app_user'@'%';
-GRANT SELECT ON flags.* TO 'app_user'@'%';
+-- Create and grant application DB user (used by PHP web app)
+CREATE USER IF NOT EXISTS '{{ db_user | default('svc_db') }}'@'%' IDENTIFIED WITH mysql_native_password BY '{{ db_password | default('SvcDb!401') }}';
+GRANT SELECT, INSERT, UPDATE ON {{ db_name | default('referral_db') }}.* TO '{{ db_user | default('svc_db') }}'@'%';
+GRANT SELECT ON flags.* TO '{{ db_user | default('svc_db') }}'@'%';
 
 FLUSH PRIVILEGES;

tests/test_builder.py

@@ -196,6 +196,81 @@ async def test_template_builder_uses_manifest_company_context(tier1_manifest):
     assert company["name"] in spec.files["web:/var/www/portal/index.php"]
 
 
+@pytest.mark.asyncio
+async def test_template_builder_credential_reuse_steps_use_runtime_contract(tier1_manifest):
+    from open_range.builder.builder import TemplateOnlyBuilder
+
+    manifest = {
+        **tier1_manifest,
+        "bug_families": ["credential_reuse"],
+        "difficulty": {**tier1_manifest.get("difficulty", {}), "min_vulns": 1, "max_vulns": 1},
+    }
+    spec = await TemplateOnlyBuilder().build(manifest, BuildContext(seed=3, tier=1))
+
+    runtime = spec.topology.get("runtime_contract", {})
+    assert runtime
+    joined = "\n".join(step.command for step in spec.golden_path)
+    assert runtime["ldap_bind_dn"] in joined
+    assert runtime["credential_reuse_user"] in joined
+    assert runtime["credential_reuse_host"] in joined
+    assert "cn=webapp,dc=corp,dc=local" not in joined
+    assert "Svc!Ldap2024" not in joined
+
+    config_key = f"{runtime['web_host']}:{runtime['web_config_path']}"
+    config_payload = spec.files[config_key]
+    assert runtime["db_user"] in config_payload
+    assert runtime["db_password"] in config_payload
+    assert runtime["ldap_bind_dn"] in config_payload
+    assert runtime["ldap_bind_pw"] in config_payload
+
+
+def test_render_payloads_use_runtime_contract_paths_and_db_name():
+    from open_range.builder.builder import render_template_payloads
+    from open_range.protocols import TruthGraph, Vulnerability
+
+    snapshot = SnapshotSpec(
+        topology={
+            "tier": 1,
+            "hosts": ["frontend", "database", "directory"],
+            "org_name": "OpenRange",
+            "domain": "corp.local",
+            "runtime_contract": {
+                "domain": "corp.local",
+                "web_host": "frontend",
+                "db_host": "database",
+                "ldap_host": "directory",
+                "web_doc_root": "/srv/http/portal",
+                "web_config_path": "/srv/http/config.php",
+                "db_name": "clinic_db",
+                "db_user": "svc_portal",
+                "db_password": "SvcPortal!123",
+                "ldap_bind_dn": "cn=svc_portal,dc=corp,dc=local",
+                "ldap_bind_pw": "SvcPortal!123",
+                "ldap_search_base_dn": "dc=corp,dc=local",
+                "credential_reuse_user": "svc_portal",
+                "credential_reuse_host": "database",
+                "credential_reuse_password": "SvcPortal!123",
+            },
+        },
+        truth_graph=TruthGraph(
+            vulns=[
+                Vulnerability(id="v1", type="path_traversal", host="frontend"),
+                Vulnerability(id="v2", type="idor", host="frontend"),
+            ]
+        ),
+        flags=[
+            FlagSpec(id="f1", value="FLAG{path}", path="/var/flags/path_flag.txt", host="frontend"),
+            FlagSpec(id="f2", value="FLAG{db}", path="db:flags.secrets.flag", host="database"),
+        ],
+        golden_path=[],
+    )
+    files = render_template_payloads(snapshot)
+    assert "frontend:/srv/http/portal/index.php" in files
+    download = files["frontend:/srv/http/portal/download.php"]
+    assert 'readfile("/srv/http/config.php")' in download
+    assert "GRANT SELECT ON clinic_db.* TO 'leaked_user'@'%';" in files["db:sql"]
+
+
 @pytest.mark.asyncio
 async def test_template_builder_output_is_manifest_canonicalized(tier1_manifest):
     from open_range.builder.builder import TemplateOnlyBuilder

tests/test_renderer.py

@@ -362,14 +362,15 @@ def test_init_sql_creates_referral_db(renderer, sqli_spec):
         assert "billing" in sql
 
 
-def test_init_sql_grants_app_user(renderer, db_flag_spec):
-    """Template grants privileges to app_user."""
+def test_init_sql_grants_runtime_db_user(renderer, db_flag_spec):
+    """Template grants privileges to the runtime-selected DB account."""
     with tempfile.TemporaryDirectory() as tmpdir:
         out = Path(tmpdir) / "out"
         renderer.render(db_flag_spec, out)
         sql = (out / "init.sql").read_text()
         assert "GRANT" in sql
-        assert "app_user" in sql
+        assert "TO '" in sql
+        assert "app_user" not in sql
 
 
 def test_init_sql_no_file_flag(renderer, sqli_spec):

tests/test_renderer_edge_cases.py

@@ -121,7 +121,7 @@ class TestNoUsers:
             assert "useradd" not in dockerfile
 
     def test_context_defaults_db_user(self):
-        """With no users, _find_db_user should return 'app_user'."""
+        """With no users, context should synthesize a service DB account."""
         spec = SnapshotSpec(
             topology=_minimal_topology(users=[]),
             truth_graph=TruthGraph(vulns=[]),
@@ -129,8 +129,9 @@ class TestNoUsers:
             golden_path=[],
         )
         ctx = _build_context(spec)
-        assert ctx["db_user"] == "app_user"
-        assert ctx["db_pass"] == "AppUs3r!2024"
+        assert ctx["db_user"] == "svc_db"
+        assert ctx["db_pass"]
+        assert ctx["db_pass"] != "AppUs3r!2024"
 
 
 # ---------------------------------------------------------------------------
@@ -612,8 +613,8 @@ class TestDBUserResolution:
             golden_path=[],
         )
         ctx = _build_context(spec)
-        # Should fall back to default since only admin user has db access
-        assert ctx["db_user"] == "app_user"
+        # Should use synthesized service account since only admin has db access.
+        assert ctx["db_user"] == "svc_db"
 
     def test_non_admin_db_user_picked(self):
         """Non-admin user with db access should be picked."""

tests/test_renderer_integration.py

@@ -220,7 +220,7 @@ class TestDockerCompose:
 
 
 class TestInitSQL:
-    """Verify rendered init.sql has referral_db and app_user."""
+    """Verify rendered init.sql has referral_db and runtime-selected DB grants."""
 
     def test_creates_referral_db(self, rendered_dir):
         sql = (rendered_dir / "init.sql").read_text()
@@ -242,10 +242,10 @@ class TestInitSQL:
         assert "patient_referrals" in sql
         assert "billing" in sql
 
-    def test_grants_app_user(self, rendered_dir):
+    def test_grants_runtime_db_user(self, rendered_dir):
         sql = (rendered_dir / "init.sql").read_text()
-        assert "app_user" in sql
         assert "GRANT" in sql
+        assert "TO '" in sql
 
     def test_has_flush_privileges(self, rendered_dir):
         sql = (rendered_dir / "init.sql").read_text()