feat(nh3-pve): AMT static on nh3-mgmt with KVM enabled and opt-in none; WS-Man helper
- AMT re-IP'd over WS-Man to static 10.100.250.61/24 (gw/DNS .1): AMT keeps its old DHCP lease across a VLAN move, so it was unreachable until changed (reached via a temporary on-link /32 on vmbr0 + ssh tunnel, since removed). - KVM redirection enabled, redirection listener on, IPS_OptInService OptInRequired=0; all read back. Password vaulted as nh3-pve/amt-admin. - scripts/amt-wsman.py: stdlib WS-Man get/put/invoke client (digest auth, AMT 16 legacy-renegotiation TLS), password from $AMT_PW.
This commit is contained in:
Executable
+77
@@ -0,0 +1,77 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Minimal Intel AMT WS-Man client, stdlib only (no wsmancli / MeshCommander).
|
||||
|
||||
The password comes from $AMT_PW, which you fill from the vault; it is never
|
||||
printed:
|
||||
export AMT_PW="$(secret get nh3-pve/amt-admin)"
|
||||
scripts/amt-wsman.py 10.100.250.61:16993 get AMT_EthernetPortSettings \
|
||||
"InstanceID=Intel(r) AMT Ethernet Port Settings 0"
|
||||
scripts/amt-wsman.py <host:port> get IPS_OptInService
|
||||
scripts/amt-wsman.py <host:port> put <Class> <body.xml> [Selector=Value]
|
||||
scripts/amt-wsman.py <host:port> invoke <Class> <Method> <body.xml> [Selector=Value]
|
||||
|
||||
A Put body is the full instance, fields in the order Get returns them. AMT 16
|
||||
needs TLS legacy renegotiation (handled below). Used 2026-09-25 to re-IP nh3-pve's
|
||||
AMT to static 10.100.250.61, enable KVM (CIM_KVMRedirectionSAP RequestStateChange
|
||||
2), switch the redirection listener on, and set OptInRequired=0. See
|
||||
servers/nh3-pve/README.md.
|
||||
"""
|
||||
import os, ssl, sys, urllib.request, uuid, re
|
||||
|
||||
AMT = "http://intel.com/wbem/wscim/1/amt-schema/1/"
|
||||
IPS = "http://intel.com/wbem/wscim/1/ips-schema/1/"
|
||||
CIM = "http://schemas.dmtf.org/wbem/wscim/1/cim-schema/2/"
|
||||
|
||||
|
||||
def uri(cls):
|
||||
return (AMT if cls.startswith("AMT_") else IPS if cls.startswith("IPS_") else CIM) + cls
|
||||
|
||||
|
||||
def call(hostport, action, cls, body="", selectors=None, action_uri=None):
|
||||
sel = ""
|
||||
if selectors:
|
||||
sel = "<w:SelectorSet>" + "".join(f'<w:Selector Name="{k}">{v}</w:Selector>' for k, v in selectors.items()) + "</w:SelectorSet>"
|
||||
env = f"""<?xml version="1.0" encoding="utf-8"?>
|
||||
<Envelope xmlns="http://www.w3.org/2003/05/soap-envelope" xmlns:a="http://schemas.xmlsoap.org/ws/2004/08/addressing" xmlns:w="http://schemas.dmtf.org/wbem/wsman/1/wsman.xsd">
|
||||
<Header><a:Action>{action_uri or "http://schemas.xmlsoap.org/ws/2004/09/transfer/" + action}</a:Action><a:To>/wsman</a:To>
|
||||
<w:ResourceURI>{uri(cls)}</w:ResourceURI><a:MessageID>uuid:{uuid.uuid4()}</a:MessageID>
|
||||
<a:ReplyTo><a:Address>http://schemas.xmlsoap.org/ws/2004/08/addressing/role/anonymous</a:Address></a:ReplyTo>
|
||||
<w:OperationTimeout>PT60S</w:OperationTimeout>{sel}</Header><Body>{body}</Body></Envelope>"""
|
||||
ctx = ssl.create_default_context()
|
||||
ctx.check_hostname = False
|
||||
ctx.verify_mode = ssl.CERT_NONE
|
||||
ctx.options |= 0x4 # OP_LEGACY_SERVER_CONNECT: AMT 16 needs legacy renegotiation
|
||||
ctx.set_ciphers("DEFAULT@SECLEVEL=0")
|
||||
url = f"https://{hostport}/wsman"
|
||||
pm = urllib.request.HTTPPasswordMgrWithDefaultRealm()
|
||||
pm.add_password(None, url, "admin", os.environ["AMT_PW"])
|
||||
op = urllib.request.build_opener(urllib.request.HTTPSHandler(context=ctx), urllib.request.HTTPDigestAuthHandler(pm))
|
||||
req = urllib.request.Request(url, env.encode(), {"Content-Type": "application/soap+xml;charset=UTF-8"})
|
||||
return op.open(req, timeout=60).read().decode()
|
||||
|
||||
|
||||
def fields(xml):
|
||||
m = re.search(r"<(?:\w+:)?Body>(.*)</(?:\w+:)?Body>", xml, re.S)
|
||||
inner = m.group(1) if m else xml
|
||||
return re.findall(r"<(?:\w+:)?(\w+)>([^<]*)</(?:\w+:)?\1>", inner)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
hp, verb, cls = sys.argv[1:4]
|
||||
rest = sys.argv[4:]
|
||||
if verb == "get":
|
||||
sel = dict(a.split("=", 1) for a in rest) or None
|
||||
for k, v in fields(call(hp, "Get", cls, selectors=sel)):
|
||||
print(f"{k}={v}")
|
||||
elif verb == "invoke":
|
||||
method, body = rest[0], open(rest[1]).read()
|
||||
sel = dict(a.split("=", 1) for a in rest[2:]) or None
|
||||
out = call(hp, None, cls, body=body, selectors=sel, action_uri=uri(cls) + "/" + method)
|
||||
for k, v in fields(out):
|
||||
print(f"{k}={v}")
|
||||
elif verb == "put":
|
||||
body = open(rest[0]).read()
|
||||
sel = dict(a.split("=", 1) for a in rest[1:]) or None
|
||||
out = call(hp, "Put", cls, body=body, selectors=sel)
|
||||
for k, v in fields(out):
|
||||
print(f"{k}={v}")
|
||||
Reference in New Issue
Block a user