driver: digitalloggers_restapi: enable REST API

The legacy HTTP API does not work on newer units without manually
changing the configuration in Setup.

This driver is based on https://www.digital-loggers.com/restapi.pdf

Curl examples that were used for development are included as comments
so that users can test their access outside of labgrid.

The REST API seems to only allow authenticated users, so the host:
parameter is parsed to pass user and password to HTTPDigestAuth.
CSRF is also required, so a valid (simple) header is provided.
Non-authenticated URLs are supported, but most likely will not
work.

HTTPS is recommended, but the units ship with self-signed certificates
so SSL certificate verification warnings are intentionally ignored.

Example usage in lg-env.yaml (default as-shipped settings):

NetworkPowerPort:
    model: 'digitalloggers_restapi'
    host: 'http://admin:[email protected]'
    index: 0

Signed-off-by: Tim Orling <[email protected]>

doc/configuration.rst

@@ -163,6 +163,14 @@ Currently available are:
   host argument must include the protocol, such as
   ``http://192.168.0.3`` or ``http://admin:[email protected]``.
 
+``digitalloggers_restapi``
+  Controls *Digital Loggers PDUs* that use the REST API. Note that
+  host argument must include the protocol, such as
+  ``http://192.168.0.3`` or ``https://admin:[email protected]``.
+  By default, only authenticated users may access the REST API.
+  HTTPS queries intentially ignore ssl certificate validation, since
+  the as-shipped certificate is self-signed.
+
 ``eaton``
   Controls *Eaton ePDUs* via SNMP.
 

labgrid/driver/power/digitalloggers_restapi.py

@@ -0,0 +1,72 @@
+'''
+Driver for Digital Loggers PDU that use the REST API.
+Tested with Ethernet Power Controller 7.
+
+Based on https://www.digital-loggers.com/restapi.pdf
+
+By default, only an authenticated user is allowed by REST API.
+
+NetworkPowerPort:
+    model: 'digitalloggers_restapi'
+    host: 'http://admin:[email protected]'
+    index: 0
+'''
+
+import json
+import requests
+from requests.auth import HTTPDigestAuth
+from requests.packages import urllib3
+from urllib.parse import urlparse
+
+
+def extract_user_password_from_host(host):
+    url = urlparse(host)
+    if '@' in url.netloc:
+        user=url.username
+        password=url.password
+        _host= f'{url.scheme}://{url.netloc.split("@")[1]}'
+    else:
+        user = None
+        password = None
+        _host= f'{url.scheme}://{url.netloc}'
+    return user, password, _host
+
+def power_set(host, port, index, value):
+    # curl -u admin:1234 -v -X PUT -H "X-CSRF: x" --data 'value=true' --digest http://192.168.0.100/restapi/relay/outlets/=0/state/
+    # curl -u admin:1234 -v -X PUT -H "X-CSRF: x" --data 'value=false' --digest http://192.168.0.100/restapi/relay/outlets/=0/state/
+    # curl --insecure -u admin:1234 -v -X PUT -H "X-CSRF: x" --data 'value=true' --digest https://192.168.0.100/restapi/relay/outlets/=0/state/
+    # curl --insecure -u admin:1234 -v -X PUT -H "X-CSRF: x" --data 'value=false' --digest https://192.168.0.100/restapi/relay/outlets/=0/state/
+    assert port is None
+
+    index = int(index)
+    value = 'true' if value else 'false'
+    payload = {'value' : value }
+    headers = {'X-CSRF': 'x', 'Accept': 'application/json'}
+    user, password, url = extract_user_password_from_host(host)
+    host = f'{url}/restapi/relay/outlets/={index}/state/'
+    with urllib3.warnings.catch_warnings():
+        urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
+        if user and password:
+            r = requests.put(host, data=payload, auth=HTTPDigestAuth(user, password), headers=headers, verify=False)
+        else:
+            r = requests.put(host, data=payload, headers=headers, verify=False)
+        r.raise_for_status()
+
+def power_get(host, port, index):
+    # curl -u admin:1234 -v -X GET -H "X-CSRF: x" --digest http://192.168.0.100/restapi/relay/outlets/=0/state/
+    # curl --insecure -u admin:1234 -v -X GET -H "X-CSRF: x" --digest https://192.168.0.100/restapi/relay/outlets/=0/state/
+    assert port is None
+
+    index = int(index)
+    user, password, url = extract_user_password_from_host(host)
+    headers = {'X-CSRF': 'x', 'Accept': 'application/json'}
+    host = f'{url}/restapi/relay/outlets/={index}/state/'
+    with urllib3.warnings.catch_warnings():
+        urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
+        if user and password:
+            r = requests.get(host, auth=HTTPDigestAuth(user, password), headers=headers, verify=False)
+        else:
+            r = requests.get(host, headers=headers, verify=False)
+        r.raise_for_status()
+    statuses = r.json()
+    return statuses[0]