import json
import os
import sys
from html import escape as html_escape
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen
GHOST_URL = "http://ghost:2368"
ALLOWED_EMAIL = os.environ["OPS_ALLOWED_EMAIL"].strip().lower()
CONTENT_API_KEY = os.environ["GHOST_CONTENT_API_KEY"].strip()
class Handler(BaseHTTPRequestHandler):
def log_safe(self, **fields):
print(json.dumps(fields, sort_keys=True), file=sys.stderr, flush=True)
def redirect_to_ghost_404(self):
self.send_response(302)
self.send_header("Location", "/404/")
self.send_header("Cache-Control", "no-store")
self.end_headers()
def do_GET(self):
if self.path not in ("/ops", "/ops/"):
self.redirect_to_ghost_404()
return
headers = {
"Host": "news.sherlockhomeless.net",
"X-Forwarded-Proto": "https",
"User-Agent": self.headers.get("User-Agent", "ghost-ops-gate"),
}
cookie = self.headers.get("Cookie")
if cookie:
headers["Cookie"] = cookie
cookie_names = []
if cookie:
cookie_names = [part.split("=", 1)[0].strip() for part in cookie.split(";") if "=" in part]
try:
member_req = Request(f"{GHOST_URL}/members/api/member/", headers=headers)
with urlopen(member_req, timeout=10) as member_response:
self.log_safe(path=self.path, cookie_names=cookie_names, member_status=member_response.status)
if member_response.status != 200:
self.redirect_to_ghost_404()
return
member_payload = json.loads(member_response.read().decode("utf-8") or "{}")
member = member_payload.get("member") or member_payload
self.log_safe(path=self.path, member_payload_keys=sorted(member_payload.keys()), member_keys=sorted(member.keys()))
member_email = str(member.get("email", "")).lower()
self.log_safe(path=self.path, member_email_present=bool(member_email), member_email_match=member_email == ALLOWED_EMAIL)
if member_email != ALLOWED_EMAIL:
self.redirect_to_ghost_404()
return
req = Request(
f"{GHOST_URL}/ghost/api/content/posts/"
f"?key={CONTENT_API_KEY}"
"&filter=tag:hash-private-ops"
"&fields=title,slug,url,published_at,custom_excerpt"
"&order=published_at%20desc"
"&limit=50",
headers=headers,
)
with urlopen(req, timeout=10) as response:
posts_payload = json.loads(response.read().decode("utf-8") or "{}")
except (HTTPError, URLError, TimeoutError, json.JSONDecodeError):
self.redirect_to_ghost_404()
return
posts = posts_payload.get("posts") or []
if not posts:
self.redirect_to_ghost_404()
return
items = "\n".join(
f''
f'{html_escape(post["title"])}'
f''
f'{html_escape((post.get("published_at") or "")[:10])}
'
for post in posts
)
body = f"""
Private Ops
Private Ops
""".encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Cache-Control", "private, no-store")
self.end_headers()
self.wfile.write(body)
def log_message(self, format, *args):
return
if __name__ == "__main__":
ThreadingHTTPServer(("0.0.0.0", 8080), Handler).serve_forever()