ISMAIL MASSERAN
305 lines
10 KiB
Python
305 lines
10 KiB
Python
"""
|
|
Local MyKad helper API for Laravel (and browser) integration.
|
|
|
|
Binds to 127.0.0.1 only. The web app calls this from the counter PC.
|
|
|
|
Usage:
|
|
python mykad_api.py
|
|
python mykad_api.py --port 17321
|
|
|
|
Test:
|
|
curl http://127.0.0.1:17321/status
|
|
curl -X POST http://127.0.0.1:17321/read
|
|
open http://127.0.0.1:17321/demo
|
|
|
|
Optional token:
|
|
MYKAD_API_TOKEN=secret python mykad_api.py
|
|
curl -H "Authorization: Bearer secret" http://127.0.0.1:17321/read
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import json
|
|
import os
|
|
import sys
|
|
import threading
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from urllib.parse import parse_qs, urlparse
|
|
|
|
from API import list_readers
|
|
from mykad import MyKadError, read_mykad_from_reader
|
|
|
|
HELPER_NAME = "mykad-helper"
|
|
HELPER_VERSION = "1.0.0"
|
|
DEFAULT_HOST = "127.0.0.1"
|
|
DEFAULT_PORT = 17321
|
|
|
|
READ_LOCK = threading.Lock()
|
|
CONFIG = {
|
|
"token": "",
|
|
}
|
|
|
|
|
|
def _json_bytes(payload: dict) -> bytes:
|
|
return json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
|
|
|
|
|
def _data_to_api(data, include_photo: bool) -> dict:
|
|
payload = data.as_dict()
|
|
has_photo = bool(data.photo_jpeg)
|
|
payload["has_photo"] = has_photo
|
|
if include_photo and has_photo:
|
|
payload["photo_base64"] = base64.b64encode(data.photo_jpeg).decode("ascii")
|
|
payload["photo_mime"] = "image/jpeg"
|
|
else:
|
|
payload["photo_base64"] = None
|
|
payload["photo_mime"] = None
|
|
return payload
|
|
|
|
|
|
def _truthy(value) -> bool:
|
|
if value is None:
|
|
return True
|
|
if isinstance(value, bool):
|
|
return value
|
|
return str(value).strip().lower() not in ("0", "false", "no", "off")
|
|
|
|
|
|
DEMO_PAGE = """<!DOCTYPE html>
|
|
<html lang="en">
|
|
<head>
|
|
<meta charset="utf-8">
|
|
<title>MyKad helper demo</title>
|
|
<style>
|
|
body { font-family: sans-serif; max-width: 720px; margin: 2rem auto; color: #111; }
|
|
button { font-size: 1rem; padding: 0.5rem 1rem; }
|
|
pre { background: #f4f4f4; padding: 1rem; overflow: auto; }
|
|
img { max-height: 220px; border: 1px solid #ccc; }
|
|
.err { color: #a40000; }
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<h1>MyKad helper</h1>
|
|
<p>Local API demo. Insert a MyKad, then read.</p>
|
|
<p>
|
|
<button id="status">GET /status</button>
|
|
<button id="read">POST /read</button>
|
|
</p>
|
|
<p><img id="photo" alt="" hidden></p>
|
|
<pre id="out">Ready.</pre>
|
|
<script>
|
|
const out = document.getElementById("out");
|
|
const photo = document.getElementById("photo");
|
|
async function call(path, options) {
|
|
out.textContent = "Calling " + path + "...";
|
|
photo.hidden = true;
|
|
try {
|
|
const res = await fetch(path, options);
|
|
const json = await res.json();
|
|
out.textContent = JSON.stringify(json, null, 2);
|
|
if (json.ok && json.data && json.data.photo_base64) {
|
|
photo.src = "data:" + json.data.photo_mime + ";base64," + json.data.photo_base64;
|
|
photo.hidden = false;
|
|
}
|
|
} catch (err) {
|
|
out.classList.add("err");
|
|
out.textContent = String(err);
|
|
}
|
|
}
|
|
document.getElementById("status").onclick = () => call("/status");
|
|
document.getElementById("read").onclick = () => call("/read", { method: "POST" });
|
|
</script>
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
|
|
class MyKadAPIHandler(BaseHTTPRequestHandler):
|
|
def log_message(self, format, *args):
|
|
sys.stderr.write("%s - %s\n" % (self.address_string(), format % args))
|
|
|
|
def _allowed_host(self) -> bool:
|
|
host = (self.headers.get("Host") or "").split(":")[0].strip().lower()
|
|
return host in ("127.0.0.1", "localhost", "localhost.", "::1", "")
|
|
|
|
def _cors(self):
|
|
origin = self.headers.get("Origin") or "*"
|
|
self.send_header("Access-Control-Allow-Origin", origin)
|
|
self.send_header("Vary", "Origin")
|
|
self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
|
|
self.send_header(
|
|
"Access-Control-Allow-Headers",
|
|
"Content-Type, Authorization, X-MyKad-Token",
|
|
)
|
|
self.send_header("Access-Control-Max-Age", "600")
|
|
|
|
def _send(self, status: int, payload: dict | str, content_type: str = "application/json"):
|
|
if isinstance(payload, str):
|
|
body = payload.encode("utf-8")
|
|
else:
|
|
body = _json_bytes(payload)
|
|
self.send_response(status)
|
|
self._cors()
|
|
self.send_header("Content-Type", f"{content_type}; charset=utf-8")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.send_header("Cache-Control", "no-store")
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def _unauthorized(self) -> bool:
|
|
expected = CONFIG["token"]
|
|
if not expected:
|
|
return False
|
|
header = self.headers.get("Authorization") or ""
|
|
token = ""
|
|
if header.lower().startswith("bearer "):
|
|
token = header[7:].strip()
|
|
if not token:
|
|
token = (self.headers.get("X-MyKad-Token") or "").strip()
|
|
if token != expected:
|
|
self._send(401, {"ok": False, "error": "Unauthorized. Set Authorization: Bearer <token>."})
|
|
return True
|
|
return False
|
|
|
|
def do_OPTIONS(self):
|
|
if not self._allowed_host():
|
|
self._send(403, {"ok": False, "error": "Host not allowed."})
|
|
return
|
|
self.send_response(204)
|
|
self._cors()
|
|
self.end_headers()
|
|
|
|
def do_GET(self):
|
|
self._dispatch("GET")
|
|
|
|
def do_POST(self):
|
|
self._dispatch("POST")
|
|
|
|
def _read_json_body(self) -> dict:
|
|
length = int(self.headers.get("Content-Length") or 0)
|
|
if length <= 0:
|
|
return {}
|
|
raw = self.rfile.read(length)
|
|
if not raw:
|
|
return {}
|
|
try:
|
|
parsed = json.loads(raw.decode("utf-8"))
|
|
except (ValueError, UnicodeDecodeError):
|
|
raise MyKadError("Request body must be JSON.")
|
|
if not isinstance(parsed, dict):
|
|
raise MyKadError("Request body must be a JSON object.")
|
|
return parsed
|
|
|
|
def _dispatch(self, method: str):
|
|
if not self._allowed_host():
|
|
self._send(403, {"ok": False, "error": "This helper only accepts localhost requests."})
|
|
return
|
|
|
|
parsed = urlparse(self.path)
|
|
path = parsed.path.rstrip("/") or "/"
|
|
query = parse_qs(parsed.query)
|
|
|
|
if path == "/demo" and method == "GET":
|
|
self._send(200, DEMO_PAGE, "text/html")
|
|
return
|
|
|
|
if path == "/" and method == "GET":
|
|
self._send(
|
|
200,
|
|
{
|
|
"ok": True,
|
|
"service": HELPER_NAME,
|
|
"version": HELPER_VERSION,
|
|
"endpoints": {
|
|
"GET /status": "Helper health and attached readers",
|
|
"GET|POST /read": "Read the inserted MyKad",
|
|
"GET /demo": "Browser test page",
|
|
},
|
|
},
|
|
)
|
|
return
|
|
|
|
if self._unauthorized():
|
|
return
|
|
|
|
if path == "/status" and method == "GET":
|
|
readers = [str(item) for item in list_readers()]
|
|
self._send(
|
|
200,
|
|
{
|
|
"ok": True,
|
|
"service": HELPER_NAME,
|
|
"version": HELPER_VERSION,
|
|
"readers": readers,
|
|
"reader_count": len(readers),
|
|
"ready": bool(readers),
|
|
},
|
|
)
|
|
return
|
|
|
|
if path == "/read" and method in ("GET", "POST"):
|
|
try:
|
|
body = self._read_json_body() if method == "POST" else {}
|
|
except MyKadError as exc:
|
|
self._send(400, {"ok": False, "error": str(exc)})
|
|
return
|
|
|
|
reader_name = body.get("reader") or (query.get("reader") or [None])[0]
|
|
include_photo = _truthy(
|
|
body.get("include_photo", (query.get("include_photo") or [None])[0])
|
|
)
|
|
|
|
if not READ_LOCK.acquire(blocking=False):
|
|
self._send(409, {"ok": False, "error": "A MyKad read is already in progress."})
|
|
return
|
|
try:
|
|
data = read_mykad_from_reader(reader_name=reader_name or None)
|
|
self._send(200, {"ok": True, "data": _data_to_api(data, include_photo)})
|
|
except MyKadError as exc:
|
|
self._send(400, {"ok": False, "error": str(exc)})
|
|
except Exception as exc:
|
|
self._send(500, {"ok": False, "error": str(exc)})
|
|
finally:
|
|
READ_LOCK.release()
|
|
return
|
|
|
|
self._send(404, {"ok": False, "error": f"Unknown endpoint: {method} {path}"})
|
|
|
|
|
|
def main(argv=None) -> int:
|
|
parser = argparse.ArgumentParser(description="Local MyKad HTTP helper for Laravel.")
|
|
parser.add_argument("--host", default=DEFAULT_HOST, help="Bind address (default 127.0.0.1).")
|
|
parser.add_argument("--port", type=int, default=DEFAULT_PORT)
|
|
parser.add_argument(
|
|
"--token",
|
|
default=os.environ.get("MYKAD_API_TOKEN", ""),
|
|
help="Optional bearer token. Also reads MYKAD_API_TOKEN.",
|
|
)
|
|
args = parser.parse_args(argv)
|
|
|
|
if args.host not in ("127.0.0.1", "localhost"):
|
|
print("Refusing to bind a non-localhost address.", file=sys.stderr)
|
|
return 2
|
|
|
|
CONFIG["token"] = (args.token or "").strip()
|
|
server = ThreadingHTTPServer((args.host, args.port), MyKadAPIHandler)
|
|
print(f"{HELPER_NAME} {HELPER_VERSION} listening on http://{args.host}:{args.port}")
|
|
print("GET /status")
|
|
print("POST /read")
|
|
print("GET /demo")
|
|
if CONFIG["token"]:
|
|
print("Token authentication is ON.")
|
|
try:
|
|
server.serve_forever()
|
|
except KeyboardInterrupt:
|
|
print("\nStopped.")
|
|
finally:
|
|
server.server_close()
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|