#!/usr/bin/env python3 # Author: Tom Sapletta ยท https://tom.sapletta.com # Part of the ifURI solution. # # Local-only LinkedIn-like development surface. It has a login form backed by .env # credentials and a fake feed with a publish form. No network calls, no real social # service, no external side effects. from __future__ import annotations import argparse import html import json import secrets import threading import time import urllib.parse from dataclasses import dataclass, field from http import HTTPStatus from http.cookies import SimpleCookie from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any HERE = Path(__file__).resolve().parent DEFAULT_ENV = HERE / ".env" EXAMPLE_ENV = HERE / ".env.example" def load_env(path: str | Path = DEFAULT_ENV) -> dict[str, str]: """Small .env reader for the example; no python-dotenv dependency.""" env_path = Path(path) if not env_path.exists() and env_path.name == ".env" and EXAMPLE_ENV.exists(): env_path = EXAMPLE_ENV data: dict[str, str] = {} for raw in env_path.read_text(encoding="utf-8").splitlines() if env_path.exists() else []: line = raw.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) data[key.strip()] = value.strip().strip('"').strip("'") return data @dataclass class MockState: user: str password: str name: str sessions: set[str] = field(default_factory=set) posts: list[dict[str, Any]] = field(default_factory=list) @classmethod def from_env(cls, env: dict[str, str]) -> "MockState": return cls( user=env.get("REAL_LINKEDIN_USER") or env.get("LINKEDIN_USER") or env.get("FAKE_LINKEDIN_USER", "tom@example.local"), password=env.get("REAL_LINKEDIN_PASSWORD") or env.get("LINKEDIN_PASSWORD") or env.get("FAKE_LINKEDIN_PASSWORD", "dev-password-123"), name=env.get("REAL_LINKEDIN_NAME") or env.get("LINKEDIN_NAME") or env.get("FAKE_LINKEDIN_NAME", "Tom Developer"), ) def publish(self, content: str) -> dict[str, Any]: post = { "id": f"post-{len(self.posts) + 1}", "author": self.name, "content": content, "createdAt": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), } self.posts.insert(0, post) return post def _page(title: str, body: str) -> bytes: return f""" {html.escape(title)} {body} """.encode("utf-8") class MockLinkedInHandler(BaseHTTPRequestHandler): server_version = "LinkedIn/0.1" @property def state(self) -> MockState: return self.server.state # type: ignore[attr-defined] def log_message(self, fmt: str, *args: Any) -> None: return def _send(self, status: int, body: bytes, content_type: str = "text/html; charset=utf-8", headers: dict[str, str] | None = None) -> None: self.send_response(status) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(body))) for key, value in (headers or {}).items(): self.send_header(key, value) self.end_headers() self.wfile.write(body) def _redirect(self, location: str, headers: dict[str, str] | None = None) -> None: self._send(HTTPStatus.SEE_OTHER, b"", headers={"Location": location, **(headers or {})}) def _form(self) -> dict[str, str]: length = int(self.headers.get("Content-Length", "0") or "0") raw = self.rfile.read(length).decode("utf-8", "replace") return {k: v[0] if v else "" for k, v in urllib.parse.parse_qs(raw).items()} def _session_id(self) -> str: cookie = SimpleCookie(self.headers.get("Cookie")) return cookie.get("fake_li_session").value if cookie.get("fake_li_session") else "" def _authenticated(self) -> bool: return self._session_id() in self.state.sessions def do_GET(self) -> None: path = urllib.parse.urlparse(self.path).path if path in {"/", "/feed"}: if not self._authenticated(): self._redirect("/login?next=/feed") return self._send(200, self._feed()) return if path == "/login": self._send(200, self._login()) return if path == "/api/posts": body = json.dumps({"ok": True, "posts": self.state.posts}, ensure_ascii=False).encode("utf-8") self._send(200, body, "application/json; charset=utf-8") return self._send(404, b"not found", "text/plain; charset=utf-8") def do_POST(self) -> None: path = urllib.parse.urlparse(self.path).path if path == "/login": form = self._form() if form.get("username") == self.state.user and form.get("password") == self.state.password: sid = secrets.token_urlsafe(24) self.state.sessions.add(sid) self._redirect("/feed", {"Set-Cookie": f"fake_li_session={sid}; Path=/; HttpOnly; SameSite=Lax"}) return self._send(401, self._login("Invalid local development credentials.")) return if path == "/post": if not self._authenticated(): self._redirect("/login?next=/feed") return content = self._form().get("content", "").strip() if content: self.state.publish(content) self._redirect("/feed") return self._send(404, b"not found", "text/plain; charset=utf-8") def _login(self, error: str = "") -> bytes: err = f'

{html.escape(error)}

' if error else "" return _page("LinkedIn Login", f"""
in

Sign in to LinkedIn

Local development surface. Credentials come from .env.

{err}
""") def _feed(self) -> bytes: posts = "\n".join( f"""

{html.escape(post['author'])}

{html.escape(post['createdAt'])}

{html.escape(post['content'])}

""" for post in self.state.posts ) or '
No posts yet.
' return _page("LinkedIn Feed", f"""
in

Start a post

{posts}
""") def start_server(host: str, port: int, env_path: str | Path = DEFAULT_ENV) -> tuple[ThreadingHTTPServer, MockState]: state = MockState.from_env(load_env(env_path)) server = ThreadingHTTPServer((host, port), MockLinkedInHandler) server.state = state # type: ignore[attr-defined] threading.Thread(target=server.serve_forever, daemon=True).start() return server, state def main() -> int: parser = argparse.ArgumentParser(description="Run a local Fake LinkedIn development server.") parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--port", type=int, default=8080) parser.add_argument("--env", default=str(DEFAULT_ENV)) args = parser.parse_args() state = MockState.from_env(load_env(args.env)) server = ThreadingHTTPServer((args.host, args.port), MockLinkedInHandler) server.state = state # type: ignore[attr-defined] print(json.dumps({"ok": True, "url": f"http://{args.host}:{args.port}/feed", "user": state.user})) try: server.serve_forever() except KeyboardInterrupt: return 130 return 0 if __name__ == "__main__": raise SystemExit(main())