From fe44430a461251c0b368b0d7add785118f8a2d50 Mon Sep 17 00:00:00 2001
From: Rekushi <170186554+RekuNote@users.noreply.github.com>
Date: Mon, 31 Aug 2026 20:29:55 +0100
Subject: [PATCH] Port hatena.py to Python 3 and modern Twisted
* Replace Python 2 syntax and string handling with Python 3-compatible code
* Add explicit bytes/text conversion for Twisted requests and responses
* Update request paths, headers, arguments, and child resource names for modern Twisted
* Return byte responses from resource render methods
* Replace deprecated `imp.load_source()` with `importlib.util`
* Generate unique module names for dynamically loaded Hatena endpoints
* Update HTTP response/header handling
* Preserve Nintendo DSi connection-test handling
* Preserve Flipnote Hatena proxy access restrictions
* Preserve region routing for v2-xx, v2-eu, v2-us, and v2-jp
* Preserve UGOXML loading and packing behavior
* Update static file and folder resource handling
* Improve 404 query-string reconstruction
* Add compatibility fallback for modern Twisted client address APIs
* Keep the original dynamic `hatenadir` resource-tree behavior
---
hatena.py | 480 ++++++++++++++++++++++++++++++------------------------
1 file changed, 271 insertions(+), 209 deletions(-)
diff --git a/hatena.py b/hatena.py
index 9c4863c..cca2bf7 100644
--- a/hatena.py
+++ b/hatena.py
@@ -1,222 +1,284 @@
-from twisted.web import static, resource
-import sys, os, imp
+from __future__ import annotations
-from Hatenatools import *#UGO, PPM, TMB and NTFT
-#from DB import Database#flipnote handler
-#from DB import Database#flipnote handler
-ServerLog = None#Is set to the log class by server.py
+import hashlib
+import importlib.util
+import os
+import sys
+from urllib.parse import urlencode
+from twisted.web import resource, static
+
+from Hatenatools import NTFT, PPM, TMB, UGO
+
+ServerLog = None # Set by server.py
Silent = False
+
+
+def as_text(value):
+ if isinstance(value, bytes):
+ return value.decode("utf-8", "replace")
+ return str(value)
+
+
+def as_bytes(value):
+ if isinstance(value, bytes):
+ return value
+ return str(value).encode("utf-8")
+
+
+def request_path(request):
+ return as_text(request.path)
+
+
+def request_headers(request):
+ return {as_text(k).lower(): as_text(v) for k, v in request.getAllHeaders().items()}
+
+
+def request_args(request):
+ out = {}
+ for key, values in request.args.items():
+ out[as_text(key)] = [as_text(v) for v in values]
+ return out
+
+
+def client_ip(request):
+ # getClientIP exists on the Twisted versions this project historically used;
+ # getClientAddress is the modern fallback.
+ getter = getattr(request, "getClientIP", None)
+ if getter is not None:
+ value = getter()
+ if value:
+ return as_text(value)
+ address = request.getClientAddress()
+ return as_text(getattr(address, "host", address))
+
+
def Log(request, path=None, silent=Silent):
- if not path:
- path = "\"%s\"" % request.path
- ServerLog.write("%s requested %s" % (request.getClientIP(), path), Silent)
+ if path is None:
+ path = '"%s"' % request_path(request)
+ if ServerLog is not None:
+ ServerLog.write("%s requested %s" % (client_ip(request), path), silent)
-#Responses:
-class AccessDenied(resource.Resource):
- isLeaf = True
- def render(self, request):
- ServerLog.write("%s got 403 when requesting \"%s\"" % (request.getClientIP(), request.path), Silent)
- print "debug:",request.getAllHeaders()
-
- request.setResponseCode(403)
- return "403 - Access denied\nThis proxy is only allowed to use for Flipnote Hatena for the DSi."
-AccessDenied = AccessDenied()
-class NotFound(resource.Resource):
- isLeaf = True
- def render(self, request):
- args = "&".join(("%s=%s" % (i, request.args[i][0]) for i in request.args))
- path = ("%s?%s" % (request.path, args)) if args else request.path
-
- ServerLog.write("%s got 404 when requesting \"%s\"" % (request.getClientIP(), path), Silent)
- request.setResponseCode(404)
- return "404 - Not Found\nThis proxy is only allowed to use for Flipnote Hatena for the DSi."
-NotFound = NotFound()
-class ConnectionTest(resource.Resource):#used for people setting up the proxy in their DSi. It's nice to see it's actually working.
- #http://conntest.nintendowifi.net/
- isLeaf = True
- def render(self, request):
- ServerLog.write("%s performed a connection test" % request.getClientIP(), True)
- request.responseHeaders.setRawHeaders('X-Organization', ['Nintendo'])#i hope this isn't illegal...
- return 'HTML PageThis is test.html page'
-ConnectionTest = ConnectionTest()
-#Root structure:
-class Root(resource.Resource):#filters out non-hantena clients
- isLeaf = False
- def __init__(self):
- resource.Resource.__init__(self)
-
- self.dsResource = ds()
- self.cssResource = static.File("hatenadir/css/")
- self.imagesResource = static.File("hatenadir/images/")
- def getChild(self, name, request):
- if "x-dsi-sid" not in request.getAllHeaders():
- if "host" in request.getAllHeaders():
- if request.getAllHeaders()["host"] == "conntest.nintendowifi.net":
- return ConnectionTest
- return AccessDenied
-
- if name == "ds":
- return self.dsResource
- elif name == "css":
- return self.cssResource
- elif name == "images":
- return self.imagesResource
- elif name == "":
- return self
-
- #return NotFound
- return self
- def render(self, request):
- if request.getHost() == "conntest.nintendowifi.net":
- return ConnectionTest
- Log(request, "root")
- return "Welcome to hatena.pbsds.net!\nThis is still in early stages, so please don't expect too much."
-class ds(resource.Resource):#child of Root
- isLeaf = False
- def __init__(self):
- resource.Resource.__init__(self)
-
- self.region = UgoRoot()
- self.regions = ("v2-xx", "v2-eu", "v2-us", "v2-jp")
- def getChild(self, name, request):
- if name in self.regions:
- return self.region
- elif name == "":
- return self
- return NotFound
- def render(self, request):
- Log(request)
- return "ds desu"
-class UgoRoot(resource.Resource):#child of ds. (v2-xx)
- isLeaf = False
- def __init__(self):
- resource.Resource.__init__(self)
-
- LoadHatenadirStructure(self)
- def getChild(self, name, request):
- if name == "":
- return self
- return NotFound
- def render(self, request):
- Log(request, "ugoroot")
- return "UgoRoot desu"
+class AccessDeniedResource(resource.Resource):
+ isLeaf = True
+
+ def render(self, request):
+ if ServerLog is not None:
+ ServerLog.write(
+ '%s got 403 when requesting "%s"' % (client_ip(request), request_path(request)),
+ Silent,
+ )
+ request.setResponseCode(403)
+ request.setHeader(b"content-type", b"text/plain; charset=utf-8")
+ return b"403 - Access denied\nThis proxy is only allowed to use for Flipnote Hatena for the DSi."
+
+
+AccessDenied = AccessDeniedResource()
+
+
+class NotFoundResource(resource.Resource):
+ isLeaf = True
+
+ def render(self, request):
+ args = request_args(request)
+ query = urlencode([(k, v) for k, values in args.items() for v in values])
+ path = request_path(request)
+ if query:
+ path += "?" + query
+ if ServerLog is not None:
+ ServerLog.write('%s got 404 when requesting "%s"' % (client_ip(request), path), Silent)
+ request.setResponseCode(404)
+ request.setHeader(b"content-type", b"text/plain; charset=utf-8")
+ return b"404 - Not Found\nThis proxy is only allowed to use for Flipnote Hatena for the DSi."
+
+
+NotFound = NotFoundResource()
+
+
+class ConnectionTestResource(resource.Resource):
+ isLeaf = True
+
+ def render(self, request):
+ if ServerLog is not None:
+ ServerLog.write("%s performed a connection test" % client_ip(request), True)
+ request.setHeader(b"X-Organization", b"Nintendo")
+ request.setHeader(b"content-type", b"text/html; charset=utf-8")
+ return b'HTML PageThis is test.html page'
+
+
+ConnectionTest = ConnectionTestResource()
+
+
+class Root(resource.Resource):
+ isLeaf = False
+
+ def __init__(self):
+ super().__init__()
+ self.dsResource = ds()
+ self.cssResource = static.File("hatenadir/css/")
+ self.imagesResource = static.File("hatenadir/images/")
+
+ def getChild(self, name, request):
+ name = as_text(name)
+ headers = request_headers(request)
+ if "x-dsi-sid" not in headers:
+ if headers.get("host", "").split(":", 1)[0].lower() == "conntest.nintendowifi.net":
+ return ConnectionTest
+ return AccessDenied
+
+ if name == "ds":
+ return self.dsResource
+ if name == "css":
+ return self.cssResource
+ if name == "images":
+ return self.imagesResource
+ if name == "":
+ return self
+ # The original server intentionally tolerated extra leading path pieces.
+ return self
+
+ def render(self, request):
+ headers = request_headers(request)
+ if headers.get("host", "").split(":", 1)[0].lower() == "conntest.nintendowifi.net":
+ return ConnectionTest.render(request)
+ Log(request, "root")
+ request.setHeader(b"content-type", b"text/plain; charset=utf-8")
+ return b"Welcome to hatena.pbsds.net!\nThis is still in early stages, so please don't expect too much."
+
+
+class ds(resource.Resource):
+ isLeaf = False
+
+ def __init__(self):
+ super().__init__()
+ self.region = UgoRoot()
+ self.regions = {"v2-xx", "v2-eu", "v2-us", "v2-jp"}
+
+ def getChild(self, name, request):
+ name = as_text(name)
+ if name in self.regions:
+ return self.region
+ if name == "":
+ return self
+ return NotFound
+
+ def render(self, request):
+ Log(request)
+ return b"ds desu"
+
+
+class UgoRoot(resource.Resource):
+ isLeaf = False
+
+ def __init__(self):
+ super().__init__()
+ LoadHatenadirStructure(self)
+
+ def getChild(self, name, request):
+ if as_text(name) == "":
+ return self
+ return NotFound
+
+ def render(self, request):
+ Log(request, "ugoroot")
+ return b"UgoRoot desu"
+
-#UgoRoot filestructure:
class FileResource(resource.Resource):
- isLeaf = True
- def __init__(self, filepath, Store=False):
- resource.Resource.__init__(self)
-
- self.Store = Store
-
- if Store:
- f = open(filepath, "rb")
- self.file = f.read()
- f.close()
- else:
- self.file = filepath
-
- self.html = filepath.split(".")[-1][:3] == "htm"
- def render(self, request):
- path = "/".join(request.path.split("/")[3:])
- Log(request, path)
-
- if self.html:
- #request.responseHeaders.setRawHeaders('content-type', ['text/html'])
- pass
- else:
- request.responseHeaders.setRawHeaders('content-type', ['text/plain'])
-
- if self.Store:
- return self.file
- else:
- #f = open(self.file, "rb")
- #ret = f.read()
- #f.close()
- #return ret
-
- return static.File(self.file).render(request)
+ isLeaf = True
+
+ def __init__(self, filepath, Store=False):
+ super().__init__()
+ self.Store = Store
+ self.file = open(filepath, "rb").read() if Store else filepath
+ self.html = filepath.split(".")[-1][:3].lower() == "htm"
+
+ def render(self, request):
+ path = "/".join(request_path(request).split("/")[3:])
+ Log(request, path)
+ if self.html:
+ request.setHeader(b"content-type", b"text/html; charset=utf-8")
+ else:
+ request.setHeader(b"content-type", b"text/plain")
+ if self.Store:
+ return self.file
+ return static.File(self.file).render(request)
+
+
class UGOXMLResource(resource.Resource):
- isLeaf = True
- def __init__(self, filepath):
- resource.Resource.__init__(self)
-
-
- self.ugofile = UGO().ReadXML(filepath, False)
- self.ugofile = self.ugofile.Pack()
- def render(self, request):
- path = "/".join(request.path.split("/")[3:])
- Log(request, path)
-
- #page = int(request.args["page"][0]) if "page" in request.args else 1
- request.responseHeaders.setRawHeaders('content-type', ['text/plain'])
- return self.ugofile
+ isLeaf = True
+
+ def __init__(self, filepath):
+ super().__init__()
+ ugo = UGO().ReadXML(filepath, False)
+ if not ugo:
+ raise ValueError("Could not parse UGOXML file: %s" % filepath)
+ self.ugofile = ugo.Pack()
+
+ def render(self, request):
+ path = "/".join(request_path(request).split("/")[3:])
+ Log(request, path)
+ request.setHeader(b"content-type", b"text/plain")
+ return self.ugofile
+
+
class FolderResource(resource.Resource):
- isLeaf = False
- def getChild(self, name, request):
- if name == "":
- return self
- return NotFound
- def render(self, request):
- path = "/".join(request.path.split("/")[3:])
- Log(request, path)
- return "I am a folder, but I'm to lazy to list my contents..."
+ isLeaf = False
+
+ def getChild(self, name, request):
+ if as_text(name) == "":
+ return self
+ return NotFound
+
+ def render(self, request):
+ path = "/".join(request_path(request).split("/")[3:])
+ Log(request, path)
+ return b"I am a folder, but I'm too lazy to list my contents..."
+
+
+def _load_python_resource(filepath):
+ module_name = "hatena_dynamic_" + hashlib.sha1(os.path.abspath(filepath).encode("utf-8")).hexdigest()
+ spec = importlib.util.spec_from_file_location(module_name, filepath)
+ if spec is None or spec.loader is None:
+ raise ImportError("Could not create import spec for %s" % filepath)
+ module = importlib.util.module_from_spec(spec)
+ sys.modules[module_name] = module
+ spec.loader.exec_module(module)
+ return module
+
+
def LoadHatenadirStructure(Resource, path=os.path.join("hatenadir", "ds", "v2-xx")):
- for root, dirs, files in os.walk(path):
- if root <> path: continue#use recursion instead
- for filename in files:
- filetype = filename.split(".")[-1].lower()
- os.path.join(path, filename)
-
- if filetype == "ugoxml":
- Resource.putChild(filename[:-3], UGOXMLResource(os.path.join(path, filename)))
- elif filetype == "py":
- try:
- pyfile = imp.load_source("pyfile", os.path.abspath(os.path.join(path, filename)))
- except ImportError as err:
- pyfile = None
- print "Error!"
- print "Failed to import the python file \"%s\"" % os.path.join(path, filename)
- print err
-
- if pyfile:
- Resource.putChild(filename[:-3], pyfile.PyResource())
- elif filetype == "pyc":
- pass#ignore
- else:
- Resource.putChild(filename, FileResource(os.path.join(path, filename)))
- for foldername in dirs:# os.path.isdir(i):
- if foldername[:2] <> "__":
- folder = FolderResource()
- LoadHatenadirStructure(folder, os.path.join(path, foldername))
- Resource.putChild(foldername, folder)
+ for root, dirs, files in os.walk(path):
+ if root != path:
+ continue
+
+ for filename in files:
+ filetype = filename.split(".")[-1].lower()
+ filepath = os.path.join(path, filename)
+ if filetype == "ugoxml":
+ Resource.putChild(filename[:-3].encode("utf-8"), UGOXMLResource(filepath))
+ elif filetype == "py":
+ try:
+ pyfile = _load_python_resource(os.path.abspath(filepath))
+ except Exception as err:
+ print("Error!")
+ print('Failed to import the python file "%s"' % filepath)
+ print(err)
+ raise
+ Resource.putChild(filename[:-3].encode("utf-8"), pyfile.PyResource())
+ elif filetype == "pyc":
+ continue
+ else:
+ Resource.putChild(filename.encode("utf-8"), FileResource(filepath))
+
+ for foldername in dirs:
+ if not foldername.startswith("__"):
+ folder = FolderResource()
+ LoadHatenadirStructure(folder, os.path.join(path, foldername))
+ Resource.putChild(foldername.encode("utf-8"), folder)
+ break
+
-#when loading:
def Setup():
- root = Root()
- return root
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
+ return Root()