Add lab-02-security-as-a-system

lab-02-security-as-a-system/scenario/status_service.py

@@ -0,0 +1,108 @@
+#!/usr/bin/env python3
+
+import json
+import os
+import threading
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+from service_common import record_event
+
+
+HOST = os.environ["MINICORP_SERVICE_HOST"]
+PORT = int(os.environ["MINICORP_SERVICE_PORT"])
+
+SERVICE_ID = os.environ["MINICORP_SERVICE_ID"]
+SERVICE_TOKEN = os.environ["MINICORP_SERVICE_TOKEN"]
+
+EVIDENCE_LOG = os.environ["MINICORP_EVIDENCE_LOG"]
+
+SEQUENCE = (
+    "ok",
+    "ok",
+    "error",
+    "ok",
+    "error",
+)
+
+lock = threading.Lock()
+request_count = 0
+
+
+def next_result():
+    global request_count
+
+    with lock:
+        request_count += 1
+        number = request_count
+
+    result = SEQUENCE[
+        (number - 1) % len(SEQUENCE)
+    ]
+
+    return number, result
+
+
+class StatusHandler(BaseHTTPRequestHandler):
+    def do_GET(self):
+        if self.path != "/":
+            self.send_error(404)
+            return
+
+        number, result = next_result()
+        source = self.client_address[0]
+
+        record_event(
+            EVIDENCE_LOG,
+            service="status",
+            service_token=SERVICE_TOKEN,
+            source=source,
+            port=PORT,
+            result=result,
+            request=number,
+        )
+
+        response = {
+            "status": result,
+            "service": "minicorp-api",
+            "service_id": SERVICE_ID,
+        }
+
+        body = (
+            json.dumps(
+                response,
+                sort_keys=True,
+            )
+            + "\n"
+        ).encode("utf-8")
+
+        self.send_response(200)
+        self.send_header(
+            "Content-Type",
+            "application/json",
+        )
+        self.send_header(
+            "Content-Length",
+            str(len(body)),
+        )
+        self.end_headers()
+        self.wfile.write(body)
+
+    def log_message(self, format, *args):
+        pass
+
+
+def main():
+    server = ThreadingHTTPServer(
+        (HOST, PORT),
+        StatusHandler,
+    )
+
+    print(
+        f"MiniCorp status service listening on {HOST}:{PORT}"
+    )
+
+    server.serve_forever()
+
+
+if __name__ == "__main__":
+    main()