diff --git a/scripts/amt-cira-setup.py b/scripts/amt-cira-setup.py new file mode 100755 index 0000000..52e34b8 --- /dev/null +++ b/scripts/amt-cira-setup.py @@ -0,0 +1,187 @@ +#!/usr/bin/env python3 +"""Configure Intel AMT to phone home (CIRA) to a MeshCentral MPS, over the LAN, idempotently. + +MeshCentral's own AMT manager only pushes CIRA through a MeshAgent on the host (LMS; amtmanager.js +"Only setup CIRA when LMS connection is used"). For an agent-less AMT (nh3-pve's), this script does +the same five steps directly over WS-Man, mirroring amtmanager.js (MeshCentral 1.2.0): + + 1. trust MeshCentral's root certificate AMT_PublicKeyManagementService.AddTrustedRootCertificate + 2. add the MPS server AMT_RemoteAccessService.AddMpServer + (FQDN, port, username/password auth; the username is the first 16 chars of the device group's + meshid with @ and $ replaced by X, which is how MeshCentral files the device into the group) + 3. add a periodic CIRA policy for it AMT_RemoteAccessService.AddRemoteAccessPolicyRule + (Trigger 2 = periodic, every 10 s, TunnelLifeTime 0 = stay up; MeshCentral's own values) + 4. allow BIOS + OS initiated connections AMT_UserInitiatedConnectionService.RequestStateChange 32771 + 5. set a random environment-detection domain AMT_EnvironmentDetectionSettingData.DetectionStrings, + so AMT always considers itself "outside" and uses the tunnel + +Secrets come from the environment, never from argv: AMT_PW (AMT admin), MPS_PW (MeshCentral mpsPass). + + export AMT_PW="$(secret get nh3-pve/amt-admin | head -1)" + export MPS_PW="$(secret get pfi-tacticalrmm/meshcentral-mpspass | tr -d '\\n')" + scripts/amt-cira-setup.py 10.100.250.61:16993 --mps rmm-mesh.phasefinal.com:4433 \\ + --user CtDDEpGX0VLlJ1X9 --root-cert-b64 # read-only state report + ... --apply # make the changes + +Undo: delete the policy rule, the MPS SAP and (optionally) the trusted root via WS-Man Delete, and +clear DetectionStrings. First used 2026-10-02 for nh3-pve (Prime). +""" +import argparse, base64, importlib.util, os, re, secrets, sys, xml.etree.ElementTree as ET + +_spec = importlib.util.spec_from_file_location("amtw", os.path.join(os.path.dirname(os.path.abspath(__file__)), "amt-wsman.py")) +amtw = importlib.util.module_from_spec(_spec); _spec.loader.exec_module(amtw) +call, uri = amtw.call, amtw.uri + +ENUM = "http://schemas.xmlsoap.org/ws/2004/09/enumeration/Enumerate" +PULL = "http://schemas.xmlsoap.org/ws/2004/09/enumeration/Pull" +SAP = "AMT_ManagementPresenceRemoteSAP" + + +def local(tag): + return tag.split("}", 1)[-1] + + +def to_dict(el): + d = {} + for c in el: + v = c.text if len(c) == 0 else to_dict(c) + if local(c.tag) == "Selector": + v = (c.attrib.get("Name"), c.text) + if local(c.tag) in d: + if not isinstance(d[local(c.tag)], list): + d[local(c.tag)] = [d[local(c.tag)]] + d[local(c.tag)].append(v) + else: + d[local(c.tag)] = v + return d + + +def enum(hp, cls): + # AMT 16 ignores OptimizeEnumeration and returns only a context, so Enumerate then Pull until + # EndOfSequence (as MeshCentral's amt-wsman does). + root = ET.fromstring(call(hp, None, cls, '', action_uri=ENUM)) + ctx = [e.text for e in root.iter() if local(e.tag) == "EnumerationContext"][0] + out = [] + for _ in range(50): + body = ('' + f'{ctx}99999999') + root = ET.fromstring(call(hp, None, cls, body, action_uri=PULL)) + for items in (e for e in root.iter() if local(e.tag) == "Items"): + out.extend(to_dict(i) for i in items) + if any(local(e.tag) == "EndOfSequence" for e in root.iter()): + return out + ctx = [e.text for e in root.iter() if local(e.tag) == "EnumerationContext"][0] + raise RuntimeError(f"enumeration of {cls} did not end") + + +def get(hp, cls, selectors=None): + root = ET.fromstring(call(hp, "Get", cls, "", selectors)) + body = [e for e in root.iter() if local(e.tag) == "Body"][0] + return to_dict(body[0]) if len(body) else {} + + +def invoke(hp, cls, method, args_xml): + body = f'{args_xml}' + xml = call(hp, None, cls, body, action_uri=uri(cls) + "/" + method) + m = re.search(r"<(?:\w+:)?ReturnValue>(\d+)<", xml) + rv = int(m.group(1)) if m else None + return rv, xml + + +def selectors_of(epr): + sels = epr.get("ReferenceParameters", {}).get("SelectorSet", {}).get("Selector", []) + sels = sels if isinstance(sels, list) else [sels] + return dict(s for s in sels if isinstance(s, tuple)) + + +def state(hp, root_b64, mps_host, mps_port): + certs = enum(hp, "AMT_PublicKeyCertificate") + root_present = any(c.get("X509Certificate") == root_b64 for c in certs) + saps = enum(hp, SAP) + ours = [s for s in saps if s.get("AccessInfo") == mps_host and str(s.get("Port")) == str(mps_port) and str(s.get("InfoFormat")) == "201"] + applies = enum(hp, "AMT_RemoteAccessPolicyAppliesToMPS") + pol = [] + for a in applies: + server = selectors_of(a.get("ManagedElement", {})).get("Name") + rule = selectors_of(a.get("PolicySet", {})).get("PolicyRuleName") + pol.append((rule, server)) + uic = get(hp, "AMT_UserInitiatedConnectionService") + env = get(hp, "AMT_EnvironmentDetectionSettingData") + det = env.get("DetectionStrings") + det = [] if det is None else det if isinstance(det, list) else [det] + return {"root_present": root_present, "trusted_certs": len(certs), "mps_servers": [(s.get("Name"), s.get("AccessInfo"), s.get("Port")) for s in saps], + "our_mps": ours[0].get("Name") if ours else None, "policies": pol, "uic_state": uic.get("EnabledState"), + "env_detection": det, "env": env} + + +def main(): + ap = argparse.ArgumentParser(description=__doc__.split("\n")[0]) + ap.add_argument("amt", help="AMT host:port, e.g. 10.100.250.61:16993") + ap.add_argument("--mps", required=True, help="MPS fqdn:port") + ap.add_argument("--user", required=True, help="16-char CIRA username (meshid prefix, @/$ -> X)") + ap.add_argument("--root-cert-b64", required=True, help="file holding MeshCentral's root cert, base64 DER") + ap.add_argument("--apply", action="store_true") + a = ap.parse_args() + mps_host, mps_port = a.mps.rsplit(":", 1) + if len(a.user) != 16: + sys.exit("--user must be exactly 16 chars (MeshCentral rejects others)") + root_b64 = open(a.root_cert_b64).read().strip() + hp = a.amt + + s = state(hp, root_b64, mps_host, mps_port) + show = {k: v for k, v in s.items() if k != "env"} + print("BEFORE:", show) + if not a.apply: + return + + mps_pw = os.environ["MPS_PW"] + if not s["root_present"]: + rv, _ = invoke(hp, "AMT_PublicKeyManagementService", "AddTrustedRootCertificate", f"{root_b64}") + print("1. AddTrustedRootCertificate ->", rv); assert rv == 0, rv + else: + print("1. root certificate already trusted") + + name = s["our_mps"] + if name is None: + args = (f"{mps_host}201{mps_port}" + f"2{a.user}{mps_pw}{mps_host}") + rv, xml = invoke(hp, "AMT_RemoteAccessService", "AddMpServer", args) + m = re.search(r'Selector Name="Name">([^<]+)<', xml) + name = m.group(1) if m else None + print("2. AddMpServer ->", rv, "name", name); assert rv == 0 and name, rv + else: + print("2. MPS server already present:", name) + + if ("Periodic", name) not in s["policies"]: + ext = base64.b64encode((0).to_bytes(4, "big") + (10).to_bytes(4, "big")).decode() # periodic, every 10 s + epr = ('
http://schemas.xmlsoap.org/ws/2004/08/addressing/role/anonymous
' + '' + f'{uri(SAP)}{name}' + '
') + rv, _ = invoke(hp, "AMT_RemoteAccessService", "AddRemoteAccessPolicyRule", + f"20{ext}{epr}") + print("3. AddRemoteAccessPolicyRule (periodic 10 s) ->", rv); assert rv == 0, rv + else: + print("3. periodic policy already applies to our MPS") + + if str(s["uic_state"]) != "32771": + rv, _ = invoke(hp, "AMT_UserInitiatedConnectionService", "RequestStateChange", "32771") + print("4. UserInitiatedConnection -> BIOS & OS enabled:", rv); assert rv == 0, rv + else: + print("4. user-initiated connections already BIOS & OS enabled") + + if not s["env_detection"]: + env = s["env"]; cls = "AMT_EnvironmentDetectionSettingData" + fields = "".join(f"{v}" for k, v in env.items() if k != "DetectionStrings" and isinstance(v, str)) + body = f'{fields}{secrets.token_hex(6)}' + call(hp, "Put", cls, body, {"InstanceID": env["InstanceID"]}) + print("5. environment detection set to a random domain") + else: + print("5. environment detection already set:", s["env_detection"]) + + after = state(hp, root_b64, mps_host, mps_port) + print("AFTER:", {k: v for k, v in after.items() if k != "env"}) + + +if __name__ == "__main__": + main() diff --git a/servers/pfi-tacticalrmm/README.md b/servers/pfi-tacticalrmm/README.md index 4d118a3..0bed6b8 100644 --- a/servers/pfi-tacticalrmm/README.md +++ b/servers/pfi-tacticalrmm/README.md @@ -29,6 +29,13 @@ Monitors and manages endpoints, pushes patches, runs scripts, etc. at once (AMT 16.1.25, activated, power on), so the old-TLS worry did not apply. ⚠ The path still runs through nh3-scale (CT 107 ON nh3-pve): with nh3-pve down, this AMT is unreachable from here. KVM needs an active iGPU output: the NanoKVM serves today; fit the 1080p dummy plug before it moves. +- **Phone-home (CIRA) plumbing, 2026-10-02 (Prime go):** `settings.mpsPass` set (vaulted + `pfi-tacticalrmm/meshcentral-mpspass`; without it the MPS checked only the 16-char username). FortiGate + ana-gw: service `MeshCentral-MPS-4433`, VIP `mps-to-tacticalrmm` (38.120.12.46:4433 → 10.250.50.57) and + policy 76 (wan1→servers, accept) — 4433 verified open from the internet, 4434 closed as a control. The AMT + side is `scripts/amt-cira-setup.py`. ⚠ nh3-pve's AMT took every setting but does NOT dial out: it is on a + STATIC IP, and Intel's CIRA guidance says static IP does not work (environment detection keys on DHCP + option 15). Open decision: move it to DHCP (the UDM reservation already pins 10.100.250.61). - Before 2026-10-02: `"WANonly": true` (TacticalRMM's install default). In that mode MeshCentral SILENTLY DROPS "Add Intel AMT computer": `meshuser.js` line 2682, `if (args.wanonly == true) return;`. No error, no event. LAN-mode AMT needs `WANonly` false (hybrid) + a service restart; CIRA works in WAN mode. TacticalRMM's