xCAT/confluent
2
0
Fork 0
mirror of https://github.com/xcat2/confluent.git synced 2026-10-20 14:45:45 +00:00
Code Issues Projects Releases Wiki Activity

Port the Eaton PDU plugin to asyncio

Browse Source 
The plugin was written against the http.client based SecureHTTPConnection, and
when that went away the reference was pointed at the aiohttp WebConnection,
which shares the name and nothing else. Nothing in it could run: the transport
called an async request() without awaiting it and then reached for a
getresponse() the new class does not have, and three PDUClient methods that
were never coroutines were awaited by the entry points.

Two transports now, both local to this plugin. https is aiohttp and stays on
the event loop, since the cert verifier records new fingerprints through
tasks.spawn. http is http.client in a thread, with its own socket so it can
still ask for a smaller segment size before connect: aiohttp only takes a
socket factory from 3.12 on, newer than el9, el10, ubuntu 24.04 or Leap 16
ship. That side has no cert to verify and its credentials arrive already read,
so the thread touches nothing.

connect() establishes and authenticates, wc is just the accessor now, and
logout() no longer sends a session id it never obtained. update() reports an
unsupported element instead of raising NameError.

On the https side cookies follow aiohttp's domain rules and the one POST with
a body goes out as text/plain, where http.client replayed every cookie and
sent no content type. The http side is as before, and neither can be settled
without an Eaton PDU on the bench. Both transports were exercised against a
stand-in: login, outlet read and set, sensors, logout, and a clamped segment
size on the plaintext path.
This commit is contained in:
Markus Hilger
2026-08-10 05:46:36 +02:00
parent a53351a730
commit 1c675c5f24
1 changed files with 97 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
confluent_server/confluent/plugins/hardwaremanagement/eatonpdu.py
+97 -81
View File
@@ -12,12 +12,13 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#TODO: ASYNC asyncio conversion
import asyncio
import base64
import confluent.util as util
import confluent.messages as msg
import aiohmi.util.webclient as wc
import http.client as httplib
import http.cookies as Cookie
import re
import hashlib
import json
@@ -55,86 +56,93 @@ def answer_challenge(username, password, data):
return {'sessionKey': skey.decode('utf8'), 'szResponse': rsp.decode('utf8'), 'szResponseValue': s2rsp.decode('utf8')}
import http.client as httplib
import http.cookies as Cookie
# carried over verbatim from the transport these replace
_fixedheaders = {
'Host': 'pdu.cluster.net',
'Accept': '*/*',
'Accept-Language': 'en-US,en;q=0.9',
'Connection': 'close',
'Referer': 'http://pdu.cluster.net/setting_admin.htm',
}
# Delta PDU webserver always closes connection,
# replace conditionals with always close
class SecureWebConnection(wc.WebConnection):
"""The https transport, on the loop: the cert verifier needs it"""
def __init__(self, host, verifycallback):
wc.WebConnection.__init__(self, host, 443,
verifycallback=verifycallback)
for header in _fixedheaders:
self.set_header(header, _fixedheaders[header])
async def grab_response(self, url, body=None, method=None):
# unfollowed, as http.client left them
rsp, status, _ = await self.grab_response_with_status(
url, body, method=method, allow_redirects=False)
return rsp, status
# the device closes the connection whatever it said it would do
class WebResponse(httplib.HTTPResponse):
def _check_close(self):
return True
class WebConnection(wc.WebConnection):
class PlainWebConnection(httplib.HTTPConnection):
"""The http transport, for a PDU with TLS not turned on
http.client, because the socket wants a smaller segment size set before
connect and aiohttp only takes a socket factory from 3.12 on, which is
newer than most distros ship. Requests run in a thread, which is safe
here: no cert to verify, and the credentials arrive already read.
"""
response_class = WebResponse
def __init__(self, host, secure, verifycallback):
if secure:
port = 443
else:
port = 80
wc.WebConnection.__init__(self, host, port, verifycallback=verifycallback)
self.secure = secure
def __init__(self, host):
# http.client would otherwise wait forever
httplib.HTTPConnection.__init__(self, host, 80, timeout=60)
self.stdheaders = dict(_fixedheaders)
self.cookies = {}
async def connect(self):
if self.secure:
return super(WebConnection, self).connect()
addrinfo = (await asyncio.get_running_loop().getaddrinfo(self.host, self.port))[0]
# workaround problems of too large mtu, moderately frequent occurance
# in this space
def connect(self):
addrinfo = socket.getaddrinfo(self.host, self.port, 0,
socket.SOCK_STREAM)[0]
plainsock = socket.socket(addrinfo[0])
plainsock.settimeout(self.mytimeout)
plainsock.settimeout(self.timeout)
try:
# workaround problems of too large mtu, moderately frequent
# occurance in this space
plainsock.setsockopt(socket.IPPROTO_TCP, socket.TCP_MAXSEG, 1456)
except socket.error:
pass
plainsock.connect(addrinfo[4])
self.sock = plainsock
def getresponse(self):
try:
rsp = super(WebConnection, self).getresponse()
try:
hdrs = [x.split(':', 1) for x in rsp.msg.headers]
except AttributeError:
hdrs = rsp.msg.items()
for hdr in hdrs:
if hdr[0] == 'Set-Cookie':
c = Cookie.BaseCookie(hdr[1])
for k in c:
self.cookies[k] = c[k].value
except httplib.BadStatusLine:
self.broken = True
raise
return rsp
def request(self, method, url, body=None):
headers = {}
def _blocking_grab(self, url, body, method):
headers = dict(self.stdheaders)
if body:
headers['Content-Length'] = len(body)
cookies = []
for cookie in self.cookies:
cookies.append('{0}={1}'.format(cookie, self.cookies[cookie]))
headers['Cookie'] = ';'.join(cookies)
headers['Host'] = 'pdu.cluster.net'
headers['Accept'] = '*/*'
headers['Accept-Language'] = 'en-US,en;q=0.9'
headers['Connection'] = 'close'
headers['Referer'] = 'http://pdu.cluster.net/setting_admin.htm'
return super(WebConnection, self).request(method, url, body, headers)
if self.cookies:
headers['Cookie'] = ';'.join(
['{0}={1}'.format(k, self.cookies[k]) for k in self.cookies])
self.request(method, url, body, headers)
rsp = self.getresponse()
for hdr, value in rsp.getheaders():
if hdr.lower() == 'set-cookie':
cookie = Cookie.BaseCookie(value)
for key in cookie:
self.cookies[key] = cookie[key].value
return rsp.read(), rsp.status
def grab_response(self, url, body=None, method=None):
async def grab_response(self, url, body=None, method=None):
if method is None:
method = 'GET' if body is None else 'POST'
if body:
self.request(method, url, body)
else:
self.request(method, url)
rsp = self.getresponse()
body = rsp.read()
return body, rsp.status
return await asyncio.to_thread(self._blocking_grab, url, body, method)
_sensors_by_node = {}
def get_sensor_data(element, node, configmanager):
async def get_sensor_data(element, node, configmanager):
category, name = element[-2:]
justnames = False
readings = []
@@ -149,9 +157,9 @@ def get_sensor_data(element, node, configmanager):
if not sn or sn[1] < time.time():
gc = PDUClient(node, configmanager)
try:
sdata = gc.get_sensor_data()
sdata = await gc.get_sensor_data()
finally:
gc.logout()
await gc.logout()
_sensors_by_node[node] = [sdata, time.time() + 1]
sn = _sensors_by_node.get(node, None)
for outlet in sn[0]:
@@ -183,6 +191,11 @@ class PDUClient(object):
@property
def wc(self):
# set by connect(); only login() reads it without connecting first
return self._wc
async def connect(self):
# logging in is a coroutine, so this cannot be the wc property
if self._wc:
return self._wc
targcfg = self.configmanager.get_node_attributes(self.node,
@@ -197,18 +210,18 @@ class PDUClient(object):
verifier = util.TLSCertVerifier(
self.configmanager, self.node, 'pubkeys.tls_hardwaremanager')
try:
self._wc = WebConnection(target, secure=True, verifycallback=verifier.verify_cert)
self.login(self.configmanager)
self._wc = SecureWebConnection(target, verifier.verify_cert)
await self.login(self.configmanager)
except socket.error:
pkey = self.configmanager.get_node_attributes(self.node, 'pubkeys.tls_hardwaremanager')
pkey = pkey.get(self.node, {}).get('pubkeys.tls_hardwaremanager', {}).get('value', None)
if pkey:
raise
self._wc = WebConnection(target, secure=False, verifycallback=verifier.verify_cert)
self.login(self.configmanager)
self._wc = PlainWebConnection(target)
await self.login(self.configmanager)
return self._wc
def login(self, configmanager):
async def login(self, configmanager):
credcfg = configmanager.get_node_attributes(self.node,
['secret.hardwaremanagementuser',
'secret.hardwaremanagementpassword'],
@@ -226,7 +239,7 @@ class PDUClient(object):
raise Exception('Missing username or password')
b64user = base64.b64encode(username.encode('utf8')).decode('utf8')
b64pass = base64.b64encode(passwd.encode('utf8')).decode('utf8')
rsp = self.wc.grab_response('/config/gateway?page=cgi_authentication&login={}&_dc={}'.format(b64user, int(time.time())))
rsp = await self.wc.grab_response('/config/gateway?page=cgi_authentication&login={}&_dc={}'.format(b64user, int(time.time())))
rsp = json.loads(sanitize_json(rsp[0]))
self.sessid = rsp['data'][0]
if rsp['data'][-1] == 'password':
@@ -246,22 +259,24 @@ class PDUClient(object):
parms['szResponseValue'],
int(time.time()),
)
rsp = self.wc.grab_response(url)
rsp = await self.wc.grab_response(url)
rsp = json.loads(sanitize_json(rsp[0]))
if not rsp['success']:
raise Exception('Failed to login to device')
rsp = self.wc.grab_response('/config/gateway?page=cgi_checkUserSession&sessionId={}&_dc={}'.format(self.sessid, int(time.time())))
rsp = await self.wc.grab_response('/config/gateway?page=cgi_checkUserSession&sessionId={}&_dc={}'.format(self.sessid, int(time.time())))
def do_request(self, suburl):
wc = self.wc
async def do_request(self, suburl):
wc = await self.connect()
url = '/config/gateway?page={}&sessionId={}&_dc={}'.format(suburl, self.sessid, int(time.time()))
return wc.grab_response(url)
return await wc.grab_response(url)
def logout(self):
self.do_request('cgi_logout')
async def logout(self):
if self.sessid:
await self.do_request('cgi_logout')
self.sessid = None
def get_outlet(self, outlet):
rsp = self.do_request('cgi_pdu_outlets')
async def get_outlet(self, outlet):
rsp = await self.do_request('cgi_pdu_outlets')
data = sanitize_json(rsp[0])
data = json.loads(data)
data = data['data'][0]
@@ -271,8 +286,8 @@ class PDUClient(object):
return 'on' if outdata[3] else 'off'
return
def get_sensor_data(self):
rsp = self.do_request('cgi_pdu_outlets')
async def get_sensor_data(self):
rsp = await self.do_request('cgi_pdu_outlets')
data = sanitize_json(rsp[0])
data = json.loads(data)
data = data['data'][0]
@@ -293,8 +308,9 @@ class PDUClient(object):
sdata[outletname] = outsense
return sdata
def set_outlet(self, outlet, state):
rsp = self.do_request('cgi_pdu_outlets')
async def set_outlet(self, outlet, state):
wc = await self.connect()
rsp = await self.do_request('cgi_pdu_outlets')
data = sanitize_json(rsp[0])
data = json.loads(data)
data = data['data'][0]
@@ -303,14 +319,14 @@ class PDUClient(object):
outdata = outdata[0]
if outdata[0] == outlet:
payload = "<SET_OBJECT><OBJECT name='PDU.OutletSystem.Outlet[{}].DelayBefore{}'>0</OBJECT>".format(idx, 'Startup' if state == 'on' else 'Shutdown')
rsp = self.wc.grab_response('/config/set_object_mass.xml?sessionId={}'.format(self.sessid), payload)
rsp = await wc.grab_response('/config/set_object_mass.xml?sessionId={}'.format(self.sessid), payload)
return
idx += 1
async def retrieve(nodes, element, configmanager, inputdata):
if element[0] == 'sensors':
for node in nodes:
for res in get_sensor_data(element, node, configmanager):
async for res in get_sensor_data(element, node, configmanager):
yield res
return
elif 'outlets' not in element: