Merge pull request #30 from nielstron/log_out

Log out after accessing log_in worthy services

README.md

@@ -22,13 +22,15 @@ Setting switches and reading their manual/auto state is only possible via the BL
 ### Usage
 
 ```python
+from pyblnet import test_blnet, BLNET, BLNETWeb, BLNETDirect
+
 ip = '192.168.178.10'
 
 # Check if there is a blnet at given address
 test_blnet(ip) # -> True/False
 
 # Convenient high level interface
-blnet = BLNET(ip, timeout=5)
+blnet = BLNET(ip, password='pass', timeout=5)
 
 # Control a switch by its ID
 blnet.turn_on(10)
@@ -44,12 +46,17 @@ print(blnet.fetch())
 # note that the direct use of these modules is discouraged though
 
 # Fetch the latest data via web interface
-blnet = BLNETWeb(ip, timeout=5)
-print(blnet.read_analog_values())
-print(blnet.read_digital_values())
-
-# For publishing values
-blnet.set_digital_value('10', 'AUS')
+# Note that manual log in and log out are required
+# when not using the with statement
+with BLNETWeb(ip, password='pass', timeout=5) as blnet_session:
+    print(blnet_session.read_analog_values())
+    print(blnet_session.read_digital_values())
+
+    # For publishing values
+    blnet_session.set_digital_value('10', 'AUS')
+    # Note that without explicit log out,
+    # the BLNET will block any further web access for the next 150s
+    # this is handled automatically when using the with statement
 
 # Fetch data via the Protocol developed by TA
 blnet = BLNETDirect(ip)

pyblnet/blnet.py

@@ -80,12 +80,13 @@ def fetch(self, node=None):
             'power': {},
         }
         if self.blnet_web:
-            if node is not None:
-                self.blnet_web.set_node(node)
-            data['analog'] = self._convert_web(
-                self.blnet_web.read_analog_values())
-            data['digital'] = self._convert_web(
-                self.blnet_web.read_digital_values())
+            with self.blnet_web as blnet_session:
+                if node is not None:
+                    blnet_session.set_node(node)
+                data['analog'] = self._convert_web(
+                    blnet_session.read_analog_values())
+                data['digital'] = self._convert_web(
+                    blnet_session.read_digital_values())
         if self.blnet_direct:
             direct = self.blnet_direct.get_latest(self.max_retries)[0]
             # Override values for analog and digital as values are
@@ -124,15 +125,15 @@ def turn_auto(self, digital_id, can_node=None):
 
     def _turn(self, digital_id, value, can_node=None):
         if self.blnet_web:
-            if not self.blnet_web.log_in():
-                raise ConnectionError('Could not log in')
-            if can_node is not None:
-                if not self.blnet_web.set_node(can_node):
-                    raise ConnectionError('Could not set node')
-            if not self.blnet_web.set_digital_value(digital_id, value):
-                raise ConnectionError('Failed to set value')
-            else:
-                return True
+            with self.blnet_web as blnet_session:
+                if not blnet_session.logged_in():
+                    raise ConnectionError('Could not log in')
+                if can_node is not None:
+                    if not blnet_session.set_node(can_node):
+                        raise ConnectionError('Could not set node')
+                if not blnet_session.set_digital_value(digital_id, value):
+                    raise ConnectionError('Failed to set value')
+            return True
         else:
             raise EnvironmentError('Can\'t set values with blnet web disabled')
 

pyblnet/blnet_web.py

@@ -70,6 +70,13 @@ def __init__(self, ip, password=_def_password, timeout=5):
         self.password = password
         self._timeout = timeout
 
+    def __enter__(self):
+        self.log_in()
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.log_out()
+
     def logged_in(self):
         """
         Determines whether the object is still connected to the BLNET
@@ -153,10 +160,6 @@ def set_node(self, node):
         
         Return: Still logged in (indicating successful node change)
         """
-        # ensure to be logged in
-        if not self.log_in():
-            return False
-
         # send the request to change the node
         try:
             r = requests.get(
@@ -173,10 +176,6 @@ def read_analog_values(self):
         Reads all analog values (temperatures, speeds) from the web interface
         and returns list of quadruples of id, name, value, unit of measurement
         """
-        # ensure to be logged in
-        if not self.log_in():
-            return None
-
         try:
             r = requests.get(
                 self.ip + "/580500.htm",
@@ -219,10 +218,6 @@ def read_digital_values(self):
         and returns list of quadruples of id, name, mode (AUTO/HAND), value 
         (EIN/AUS)
         """
-        # ensure to be logged in
-        if not self.log_in():
-            return None
-
         try:
             r = requests.get(
                 self.ip + "/580600.htm",
@@ -280,10 +275,6 @@ def set_digital_value(self, digital_id, value):
             raise ValueError(
                 'Device id can\'t be larger than 15, was {}'.format(
                     digital_id))
-        # ensure to be logged in
-        if not self.log_in():
-            return False
-
         # transform input value to 'EIN' or 'AUS'
         if isinstance(value, str):
             if value.lower() == 'AUTO'.lower() or value == '3':
@@ -297,9 +288,9 @@ def set_digital_value(self, digital_id, value):
             else:
                 raise ValueError("Illegal input string {}".format(value))
         elif isinstance(value, int) and not isinstance(value, bool):
-            if value is 3 or value is 2 or value is 1:
+            if value in (1, 2, 3):
                 value = str(value)
-            elif value is 0:
+            elif value == 0:
                 value = '1'
             else:
                 raise ValueError("Illegal input integer {}".format(value))

pyblnet/tests/test_structure/blnet_mock_server.py

@@ -11,13 +11,20 @@
 from pathlib import Path
 
 SERVER_DIR = Path(__file__).parent or Path('.')
+PASSWORD = '0123'
+
+COOKIE_RANGE = (0xAAAA, 0xFFFF)
+COOKIE_NUM = COOKIE_RANGE[1] - COOKIE_RANGE[0]
+LOGIN_COOKIES = [hex(i) for i in range(*COOKIE_RANGE)]
 
 
 class BLNETServer(HTTPServer):
 
+    current_log_in_cookie = 0
+
     # Currently allowed login cookie
     # or none if no one logged in
-    logged_in = True
+    logged_in = None
     # no login necessary
     password = None
     # access to digital nodes
@@ -30,8 +37,19 @@ def set_password(self, password):
         self.password = password
         self.logged_in = None
 
-    def set_logged_in(self, cookie):
-        self.logged_in = cookie
+    def log_in(self):
+        if self.logged_in is not None:
+            return False
+        self.logged_in = LOGIN_COOKIES[self.current_log_in_cookie]
+        self.current_log_in_cookie += 1
+        self.current_log_in_cookie %= COOKIE_NUM
+        return self.logged_in
+
+    def log_out(self):
+        self.logged_in = None
+
+    def is_logged_in(self, cookie):
+        return cookie is not None and self.logged_in == cookie
 
     def set_node(self, id, value):
         self.nodes[id] = value
@@ -48,6 +66,9 @@ def unset_blocked(self):
 
 class BLNETRequestHandler(SimpleHTTPRequestHandler):
 
+    # uncomment on higher python versions for better debugging
+    # server: BLNETServer
+
     def do_GET(self):
         """
         Handle get request, but check for errors in protocol
@@ -57,11 +78,14 @@ def do_GET(self):
             self.send_error(403, "Access denied because server is blocked")
             return
         path = self.translate_path(self.path)
+        if (self.server.is_logged_in(self.headers.get("cookie"))
+                and "?blL=1" in self.requestline):
+            self.server.log_out()
         # Only access that is allowed without login is main.html
         if (not Path(path) == SERVER_DIR
                 and not Path(path) == SERVER_DIR.joinpath('main.htm')
                 and not Path(path) == SERVER_DIR.joinpath('main.html')):
-            if not self.server.logged_in:
+            if not self.server.is_logged_in(self.headers.get("cookie")):
                 self.send_error(403, "Not logged in, access denied")
                 return
             # Parse node sets
@@ -121,8 +145,11 @@ def do_POST(self):
             return
         # All checks passed? set Set-Cookie header and do_GET
         # random cookie - do not hardcode
-        self.server.set_logged_in('C1A3')
-        self.headers.add_header('Cookie', 'C1A3')
+        cookie = self.server.log_in()
+        if not cookie:
+            self.send_error(403, "Previous user did not log out")
+            return
+        self.headers.add_header('Cookie', cookie)
         self.do_GET()
 
     def translate_path(self, path):
@@ -203,9 +230,8 @@ def send_head(self):
             self.send_header("Last-Modified",
                              self.date_time_string(fs.st_mtime))
             # Addition: send cookie
-            if (self.server.logged_in is not None
-                    and self.server.logged_in == self.headers.get('Cookie')):
-                self.send_header('Set-Cookie', 'C1A3')
+            if self.server.is_logged_in(self.headers.get('Cookie')):
+                self.send_header('Set-Cookie', self.server.logged_in)
             self.end_headers()
             return f
         except: