"""
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 = """
MyKad helper demo
MyKad helper
Local API demo. Insert a MyKad, then read.
![]()
Ready.
"""
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 ."})
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())