Update Flowchart.vsd

app_client_blocking.py - shortening main and other DRY fixes
app_server.py - adding lsock close to cleanup
base line secure socket framework

app_client_blocking.py

@@ -76,58 +76,15 @@ def create_message(request):
     return message
 
 
-def main():
+def read_to_buffer(sock, length, buf : bytearray):
-    if len(sys.argv) != 5:
+    while len(buf) < length:
-        print("usage:", sys.argv[0], "<host> <port> <action> <value>")
+        data = sock.recv(4096)
-        sys.exit(1)
+        buf += data
 
-    host, port = sys.argv[1], int(sys.argv[2])
-    action, value = sys.argv[3], sys.argv[4]
-    request = create_request(action, value)
-    message = create_message(request)
 
-    addr = (host, port)
+def process_response(sock, addr, jsonheader, _recv_buffer):
-    print("starting connection to", addr)
-    sock = socket.create_connection(addr)
-    sock.settimeout(5)
-
-    print(f"sending request {repr(message)} to {addr}")
-    sock.sendall(message)
-
-    _recv_buffer = bytearray(b"")
-    def read_to_buffer(sock, length, buf : bytearray):
-        while len(buf) < length:
-            data = sock.recv(4096)
-            buf += data
-
-    # read and process protoheader (fixed length of 2 byte in network order)
-    hdrlen = 2
-    read_to_buffer(sock, hdrlen, _recv_buffer)
-
-    # process_protoheader()
-    _jsonheader_len = struct.unpack(">H", _recv_buffer[:hdrlen])[0]
-    _recv_buffer = _recv_buffer[hdrlen:]
-
-    # read and process jsonheader
-    read_to_buffer(sock, _jsonheader_len, _recv_buffer)
-
-    # process_jsonheader()
-    jsonheader = _json_decode(_recv_buffer[:_jsonheader_len], "utf-8")
-    _recv_buffer = _recv_buffer[_jsonheader_len:]
-    for reqhdr in (
-        "byteorder",
-        "content-length",
-        "content-type",
-        "content-encoding",
-    ):
-        if reqhdr not in jsonheader:
-            raise ValueError(f'Missing required header "{reqhdr}".')
-
-    # read and process response
     content_len = jsonheader["content-length"]
     read_to_buffer(sock, content_len, _recv_buffer)
-
-    # process_response():
     data = _recv_buffer[:content_len]
     _recv_buffer = _recv_buffer[content_len:]
     if jsonheader["content-type"] == "text/json":
@@ -147,6 +104,45 @@ def main():
         content = response
         print(f"got response: {repr(content)}")
 
+
+def main():
+    HDRLEN = 2
+    _recv_buffer = bytearray(b"")
+
+    if len(sys.argv) != 5:
+        print("usage:", sys.argv[0], "<host> <port> <action> <value>")
+        sys.exit(1)
+
+    host, port = sys.argv[1], int(sys.argv[2])
+    action, value = sys.argv[3], sys.argv[4]
+    request = create_request(action, value)
+    message = create_message(request)
+
+    addr = (host, port)
+    print("starting connection to", addr)
+    sock = socket.create_connection(addr)
+    sock.settimeout(5)
+
+    print(f"sending request {repr(message)} to {addr}")
+    sock.sendall(message)
+
+    # read and process protoheader (fixed length of 2 byte in network order)
+    read_to_buffer(sock, HDRLEN, _recv_buffer)
+    _jsonheader_len = struct.unpack(">H", _recv_buffer[:HDRLEN])[0]
+    _recv_buffer = _recv_buffer[HDRLEN:]
+
+    # read and process jsonheader
+    read_to_buffer(sock, _jsonheader_len, _recv_buffer)
+    jsonheader = _json_decode(_recv_buffer[:_jsonheader_len], "utf-8")
+    _recv_buffer = _recv_buffer[_jsonheader_len:]
+    for reqhdr in ("byteorder", "content-length", "content-type",
+                   "content-encoding",):
+        if reqhdr not in jsonheader:
+            raise ValueError(f'Missing required header "{reqhdr}".')
+
+    # read and process response
+    process_response(sock, addr, jsonheader, _recv_buffer)
+
     sock.shutdown(socket.SHUT_RDWR)
     sock.close()
 

app_secure_client.py

@@ -0,0 +1,151 @@
+# -*- coding: utf-8 -*-
+"""
+Created on Sun Mar 22 10:46:20 2020
+
+@author: cpan
+
+Simple client implementation with timeout feature to prevent hanging
+
+"""
+
+import sys
+import socket
+import json
+import io
+import struct
+
+
+def create_request(action, value):
+    if action == "search":
+        return dict(
+            type="text/json",
+            encoding="utf-8",
+            content=dict(action=action, value=value),
+        )
+    else:
+        return dict(
+            type="binary/custom-client-binary-type",
+            encoding="binary",
+            content=bytes(action + value, encoding="utf-8"),
+        )
+
+
+def _json_encode(obj, encoding):
+    return json.dumps(obj, ensure_ascii=False).encode(encoding)
+
+
+def _json_decode(json_bytes, encoding):
+    tiow = io.TextIOWrapper(
+        io.BytesIO(json_bytes), encoding=encoding, newline=""
+    )
+    obj = json.load(tiow)
+    tiow.close()
+    return obj
+
+
+def _create_message(*, content_bytes, content_type, content_encoding):
+    jsonheader = {
+        "byteorder": sys.byteorder,
+        "content-type": content_type,
+        "content-encoding": content_encoding,
+        "content-length": len(content_bytes),
+    }
+    jsonheader_bytes = _json_encode(jsonheader, "utf-8")
+    message_hdr = struct.pack(">H", len(jsonheader_bytes))
+    message = message_hdr + jsonheader_bytes + content_bytes
+    return message
+
+
+def create_message(request):
+    content = request["content"]
+    content_type = request["type"]
+    content_encoding = request["encoding"]
+    if content_type == "text/json":
+        req = {
+            "content_bytes": _json_encode(content, content_encoding),
+            "content_type": content_type,
+            "content_encoding": content_encoding,
+        }
+    else:
+        req = {
+            "content_bytes": content,
+            "content_type": content_type,
+            "content_encoding": content_encoding,
+        }
+    message = _create_message(**req)
+    return message
+
+
+def read_to_buffer(sock, length, buf : bytearray):
+    while len(buf) < length:
+        data = sock.recv(4096)
+        buf += data
+
+
+def process_response(sock, addr, jsonheader, _recv_buffer):
+    content_len = jsonheader["content-length"]
+    read_to_buffer(sock, content_len, _recv_buffer)
+    data = _recv_buffer[:content_len]
+    _recv_buffer = _recv_buffer[content_len:]
+    if jsonheader["content-type"] == "text/json":
+        encoding = jsonheader["content-encoding"]
+        response = _json_decode(data, encoding)
+        print("received response", repr(response), "from", addr)
+        content = response
+        result = content.get("result")
+        print(f"got result: {result}")
+    else:
+        # Binary or unknown content-type
+        response = data
+        print(
+            f'received {jsonheader["content-type"]} response from',
+            addr,
+        )
+        content = response
+        print(f"got response: {repr(content)}")
+
+
+def main():
+    HDRLEN = 2
+    _recv_buffer = bytearray(b"")
+
+    if len(sys.argv) != 5:
+        print("usage:", sys.argv[0], "<host> <port> <action> <value>")
+        sys.exit(1)
+
+    host, port = sys.argv[1], int(sys.argv[2])
+    action, value = sys.argv[3], sys.argv[4]
+    request = create_request(action, value)
+    message = create_message(request)
+
+    addr = (host, port)
+    print("starting connection to", addr)
+    sock = socket.create_connection(addr)
+    sock.settimeout(5)
+
+    print(f"sending request {repr(message)} to {addr}")
+    sock.sendall(message)
+
+    # read and process protoheader (fixed length of 2 byte in network order)
+    read_to_buffer(sock, HDRLEN, _recv_buffer)
+    _jsonheader_len = struct.unpack(">H", _recv_buffer[:HDRLEN])[0]
+    _recv_buffer = _recv_buffer[HDRLEN:]
+
+    # read and process jsonheader
+    read_to_buffer(sock, _jsonheader_len, _recv_buffer)
+    jsonheader = _json_decode(_recv_buffer[:_jsonheader_len], "utf-8")
+    _recv_buffer = _recv_buffer[_jsonheader_len:]
+    for reqhdr in ("byteorder", "content-length", "content-type",
+                   "content-encoding",):
+        if reqhdr not in jsonheader:
+            raise ValueError(f'Missing required header "{reqhdr}".')
+
+    # read and process response
+    process_response(sock, addr, jsonheader, _recv_buffer)
+
+    sock.shutdown(socket.SHUT_RDWR)
+    sock.close()
+
+
+if __name__ == '__main__':
+    main()

app_secure_server.py

@@ -0,0 +1,60 @@
+# -*- coding: utf-8 -*-
+"""
+Created on Sun Mar 22 10:43:35 2020
+
+@author: cpan
+"""
+
+import sys
+import socket
+import selectors
+import traceback
+import lib_secure_server as libserver
+
+
+sel = selectors.DefaultSelector()
+
+
+def accept_wrapper(sock):
+    conn, addr = sock.accept()  # Should be ready to read
+    print("accepted connection from", addr)
+    conn.setblocking(False)
+    message = libserver.Message(sel, conn, addr)
+    sel.register(conn, selectors.EVENT_READ, data=message)
+
+
+if len(sys.argv) != 3:
+    print("usage:", sys.argv[0], "<host> <port>")
+    sys.exit(1)
+
+host, port = sys.argv[1], int(sys.argv[2])
+lsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+# Avoid bind() exception: OSError: [Errno 48] Address already in use
+lsock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+lsock.bind((host, port))
+lsock.listen()
+print("listening on", (host, port))
+lsock.setblocking(False)
+sel.register(lsock, selectors.EVENT_READ, data=None)
+
+try:
+    while True:
+        events = sel.select(timeout=None)
+        for key, mask in events:
+            if key.data is None:
+                accept_wrapper(key.fileobj)
+            else:
+                message = key.data
+                try:
+                    message.process_events(mask)
+                except Exception:
+                    print(
+                        "main: error: exception for",
+                        f"{message.addr}:\n{traceback.format_exc()}",
+                    )
+                    message.close()
+except KeyboardInterrupt:
+    print("caught keyboard interrupt, exiting")
+finally:
+    sel.close()
+    lsock.close()

app_server.py

@@ -9,9 +9,9 @@ import sys
 import socket
 import selectors
 import traceback
-
 import libserver
 
+
 sel = selectors.DefaultSelector()
 
 
@@ -57,3 +57,4 @@ except KeyboardInterrupt:
     print("caught keyboard interrupt, exiting")
 finally:
     sel.close()
+    lsock.close()

lib_secure_server.py

@@ -0,0 +1,223 @@
+# -*- coding: utf-8 -*-
+"""
+Created on Sun Mar 22 10:44:50 2020
+
+@author: cpan
+"""
+
+import sys
+import selectors
+import json
+import io
+import struct
+
+request_search = {
+    "morpheus": "Follow the white rabbit. \U0001f430",
+    "ring": "In the caves beneath the Misty Mountains. \U0001f48d",
+    "\U0001f436": "\U0001f43e Playing ball! \U0001f3d0",
+}
+
+
+class Message:
+    def __init__(self, selector, sock, addr):
+        self.selector = selector
+        self.sock = sock
+        self.addr = addr
+        self._recv_buffer = b""
+        self._send_buffer = b""
+        self._jsonheader_len = None
+        self.jsonheader = None
+        self.request = None
+        self.response_created = False
+
+    def _set_selector_events_mask(self, mode):
+        """Set selector to listen for events: mode is 'r', 'w', or 'rw'."""
+        if mode == "r":
+            events = selectors.EVENT_READ
+        elif mode == "w":
+            events = selectors.EVENT_WRITE
+        elif mode == "rw":
+            events = selectors.EVENT_READ | selectors.EVENT_WRITE
+        else:
+            raise ValueError(f"Invalid events mask mode {repr(mode)}.")
+        self.selector.modify(self.sock, events, data=self)
+
+    def _read(self):
+        try:
+            # Should be ready to read
+            data = self.sock.recv(4096)
+        except BlockingIOError:
+            # Resource temporarily unavailable (errno EWOULDBLOCK)
+            pass
+        else:
+            if data:
+                self._recv_buffer += data
+            else:
+                raise RuntimeError("Peer closed.")
+
+    def _write(self):
+        if self._send_buffer:
+            print("sending", repr(self._send_buffer), "to", self.addr)
+            try:
+                # Should be ready to write
+                sent = self.sock.send(self._send_buffer)
+            except BlockingIOError:
+                # Resource temporarily unavailable (errno EWOULDBLOCK)
+                pass
+            else:
+                self._send_buffer = self._send_buffer[sent:]
+                # Close when the buffer is drained. The response has been sent.
+                if sent and not self._send_buffer:
+                    self.close()
+
+    def _json_encode(self, obj, encoding):
+        return json.dumps(obj, ensure_ascii=False).encode(encoding)
+
+    def _json_decode(self, json_bytes, encoding):
+        tiow = io.TextIOWrapper(
+            io.BytesIO(json_bytes), encoding=encoding, newline=""
+        )
+        obj = json.load(tiow)
+        tiow.close()
+        return obj
+
+    def _create_message(
+        self, *, content_bytes, content_type, content_encoding
+    ):
+        jsonheader = {
+            "byteorder": sys.byteorder,
+            "content-type": content_type,
+            "content-encoding": content_encoding,
+            "content-length": len(content_bytes),
+        }
+        jsonheader_bytes = self._json_encode(jsonheader, "utf-8")
+        message_hdr = struct.pack(">H", len(jsonheader_bytes))
+        message = message_hdr + jsonheader_bytes + content_bytes
+        return message
+
+    def _create_response_json_content(self):
+        action = self.request.get("action")
+        if action == "search":
+            query = self.request.get("value")
+            answer = request_search.get(query) or f'No match for "{query}".'
+            content = {"result": answer}
+        else:
+            content = {"result": f'Error: invalid action "{action}".'}
+        content_encoding = "utf-8"
+        response = {
+            "content_bytes": self._json_encode(content, content_encoding),
+            "content_type": "text/json",
+            "content_encoding": content_encoding,
+        }
+        return response
+
+    def _create_response_binary_content(self):
+        response = {
+            "content_bytes": b"First 10 bytes of request: "
+            + self.request[:10],
+            "content_type": "binary/custom-server-binary-type",
+            "content_encoding": "binary",
+        }
+        return response
+
+    def process_events(self, mask):
+        if mask & selectors.EVENT_READ:
+            self.read()
+        if mask & selectors.EVENT_WRITE:
+            self.write()
+
+    def read(self):
+        self._read()
+
+        if self._jsonheader_len is None:
+            self.process_protoheader()
+
+        if self._jsonheader_len is not None:
+            if self.jsonheader is None:
+                self.process_jsonheader()
+
+        if self.jsonheader:
+            if self.request is None:
+                self.process_request()
+
+    def write(self):
+        if self.request:
+            if not self.response_created:
+                self.create_response()
+
+        self._write()
+
+    def close(self):
+        print("closing connection to", self.addr)
+        try:
+            self.selector.unregister(self.sock)
+        except Exception as e:
+            print(
+                f"error: selector.unregister() exception for",
+                f"{self.addr}: {repr(e)}",
+            )
+
+        try:
+            self.sock.close()
+        except OSError as e:
+            print(
+                f"error: socket.close() exception for",
+                f"{self.addr}: {repr(e)}",
+            )
+        finally:
+            # Delete reference to socket object for garbage collection
+            self.sock = None
+
+    def process_protoheader(self):
+        hdrlen = 2
+        if len(self._recv_buffer) >= hdrlen:
+            self._jsonheader_len = struct.unpack(
+                ">H", self._recv_buffer[:hdrlen]
+            )[0]
+            self._recv_buffer = self._recv_buffer[hdrlen:]
+
+    def process_jsonheader(self):
+        hdrlen = self._jsonheader_len
+        if len(self._recv_buffer) >= hdrlen:
+            self.jsonheader = self._json_decode(
+                self._recv_buffer[:hdrlen], "utf-8"
+            )
+            self._recv_buffer = self._recv_buffer[hdrlen:]
+            for reqhdr in (
+                "byteorder",
+                "content-length",
+                "content-type",
+                "content-encoding",
+            ):
+                if reqhdr not in self.jsonheader:
+                    raise ValueError(f'Missing required header "{reqhdr}".')
+
+    def process_request(self):
+        content_len = self.jsonheader["content-length"]
+        if not len(self._recv_buffer) >= content_len:
+            return
+        data = self._recv_buffer[:content_len]
+        self._recv_buffer = self._recv_buffer[content_len:]
+        if self.jsonheader["content-type"] == "text/json":
+            encoding = self.jsonheader["content-encoding"]
+            self.request = self._json_decode(data, encoding)
+            print("received request", repr(self.request), "from", self.addr)
+        else:
+            # Binary or unknown content-type
+            self.request = data
+            print(
+                f'received {self.jsonheader["content-type"]} request from',
+                self.addr,
+            )
+        # Set selector to listen for write events, we're done reading.
+        self._set_selector_events_mask("w")
+
+    def create_response(self):
+        if self.jsonheader["content-type"] == "text/json":
+            response = self._create_response_json_content()
+        else:
+            # Binary or unknown content-type
+            response = self._create_response_binary_content()
+        message = self._create_message(**response)
+        self.response_created = True
+        self._send_buffer += message