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
This commit is contained in:
Rekushi authored and GitHub committed
1 parent
76a9ac99e2
commit
fe44430a46
1 file changed
+271
-209
@@ -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><head><title>HTML Page</title></head><body bgcolor="#FFFFFF">This is test.html page</body></html>'
|
||||
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><head><title>HTML Page</title></head><body bgcolor="#FFFFFF">This is test.html page</body></html>'
|
||||
|
||||
|
||||
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()
|
||||
Reference in new issue
Block a user