Nova/PyWebServer
1
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases 10 Wiki Activity

Compare commits

base: Nova:amethyst-b0001
Branches Tags
Nova:main Nova:2.0 Nova:backport-1.0-issue4
Nova:amethyst-prerel-0.1.0 Nova:amethyst-b0039 Nova:amethyst-b0001 Nova:1.4.0 Nova:1.3.0 Nova:1.2.1 Nova:1.2 Nova:1.0+u2 Nova:1.1+u2 Nova:1.1 Nova:1.0
...
compare: Nova:amethyst-b0039
Branches Tags
Nova:2.0 Nova:main Nova:backport-1.0-issue4
Nova:amethyst-prerel-0.1.0 Nova:amethyst-b0039 Nova:amethyst-b0001 Nova:1.4.0 Nova:1.3.0 Nova:1.2.1 Nova:1.2 Nova:1.0+u2 Nova:1.1+u2 Nova:1.1 Nova:1.0

2 Commits
amethyst-b ... amethyst-b

Author SHA1 Message Date
Nova
04ec2ebec1 holy fuck shit finally works!!! 
alpha build 0039
 2026-03-02 21:38:28 +01:00
Nova
36c8c95efe amethyst 0018 2026-01-30 08:41:30 +01:00
4 changed files with 181 additions and 104 deletions
Expand all files Collapse all files

11
html/index.html
View File

@@ -4,10 +4,11 @@
<title>Test page</title> <title>Test page</title>
</head> </head>
<body> <body>
<h1>Hey there!</h1> <center>
<h2>You're seeing this page because you haven't set up PyWebServer yet!</h2> <h1>Hello from Amethyst!</h1>
<h2>This page confirms that PyWebServer can read and serve files from your PC.</h2> <h2>This page confirms Amethyst can read files from your PC and serve them to your browser!</h2>
<h2>To make this go away, please edit the file `pywebsrv.conf` and edit the `directory` key to your directory of choice!</h2> <p>If you see this page and you're not the server owner, tell them they misconfigured something!</p>
<p>Here you can simulate a 404 error: <a href="/uuh">Click me for a 404 error!</a></p> <p>This server runs Amethyst Alpha Build 0039</p>
</center>
</body> </body>
</html> </html>

25
new_conf.conf
View File

@@ -1,23 +1,23 @@
# WARNING: This is an alpha spec of NSCL 2.0!! # WARNING: This is an alpha spec of NSCL 2.0!!
host example.com { host 192.168.2.196 {
directory:/home/nova/Downloads/test/html directory:/home/nova/Downloads/test/html
allowed-methods:GET allowed-methods:GET
block-ip:match-ip("192.168",2) block-ua:match("Discordbot"),match("Google")
block-ua:match("Discordbot",0),match("Google",0)
} }
host cdn.example.com { host localhost {
directory:/home/nova/Downloads/test/cdn directory:/home/nova/PyWebServer/html2
allowed-methods:GET,PUT allowed-methods:GET,PUT
block-ip:10.1.100.2 block-ip:10.1.100.2
block-ua:match("Discordbot",0) block-ua:match("Discordbot")
} }
host modem.example.com { host 192.168.1.213 {
proxy:192.168.2.254 directory:/home/nova/PyWebServer/html
key-file:/home/nova/Downloads/test/proxykey.pem allowed-methods:GET,PUT
cert-file:/home/nova/Downloads/test/proxycert.pem block-ip:10.1.100.2
block-ua:match("Discordbot")
} }
globals { globals {
@@ -25,6 +25,7 @@ globals {
https:1 https:1
port:8080 port:8080
https-port:8443 https-port:8443
global-key:/home/nova/Downloads/test/key.pem allow-localhost:1
global-cert:/home/nova/Downloads/test/cert.pem key:/home/nova/PyWebServer/ssl/key.pem
cert:/home/nova/PyWebServer/ssl/cert.pem
} }

8
pywebsrv.conf
View File

@@ -4,11 +4,11 @@
port:8080 port:8080
port-https:8443 port-https:8443
# Here you choose what directory PyWebServer looks in for files. # Here you choose what directory PyWebServer looks in for files.
directory:<Enter directory here> directory:/home/nova/PyWebServer/html
# Host defenition, what hosts you can connect via. # Host defenition, what hosts you can connect via.
# You can use FQDNs, IP-addresses and localhost, # You can use FQDNs, IP-addresses and localhost,
# Support for multiple hosts is coming. # Support for multiple hosts is coming.
host:localhost host:localhost,10.185.213.118
# Enables HTTP support. (Only enables/disables the HTTP port.) # Enables HTTP support. (Only enables/disables the HTTP port.)
http:1 http:1
# Enables HTTPS support. (Only enables/disables the HTTPS port.) # Enables HTTPS support. (Only enables/disables the HTTPS port.)
@@ -21,8 +21,8 @@ allow-localhost:1
disable-autocertgen:0 disable-autocertgen:0
# If you wish to block IP-addresses, this function is coming though. # If you wish to block IP-addresses, this function is coming though.
# block-ip:0.0.0.0,1.1.1.1,2.2.2.2 # block-ip:0.0.0.0,1.1.1.1,2.2.2.2
# If you wish to block User-Agents, this function is coming though. # If you wish to block User-Agents.
# block-ua:(NULL) block-ua:match(Discordbot),match(google)
# TEST: experimental non-defined keys go here: # TEST: experimental non-defined keys go here:
# keyfile key # keyfile key

235
pywebsrv.py
View File

@@ -56,18 +56,74 @@ except ImportError:
# ) # )
pass pass
AMETHYST_BUILD_NUMBER = "0001" AMETHYST_BUILD_NUMBER = "0039"
AMETHYST_REPO = "https://git.novacow.ch/Nova/PyWebServer/" AMETHYST_REPO = "https://git.novacow.ch/Nova/PyWebServer/"
class ConfigParser:
def __init__(self, text):
self.data = {"hosts": {}, "globals": {}}
self._parse(text)
def _parse(self, text):
lines = [
line.strip()
for line in text.splitlines()
if line.strip() and not line.strip().startswith("#")
]
current_block = None
current_name = None
for line in lines:
if line.startswith("host ") and line.endswith("{"):
current_name = line.split()[1]
self.data["hosts"][current_name] = {}
current_block = ("host", current_name)
continue
if line == "globals {":
current_block = ("globals", None)
continue
if line == "}":
current_block = None
current_name = None
continue
if ":" in line and current_block:
key, value = line.split(":", 1)
key = key.strip()
value = value.strip()
if "," in value:
value = [v.strip() for v in value.split(",")]
if current_block[0] == "host":
self.data["hosts"][current_name][key] = value
else:
self.data["globals"][key] = value
def query_config(self, key, host=None):
if host:
return self.data["hosts"].get(host, {}).get(key)
if key == "hosts":
print(f"\n\n\nHosts!\nHosts: {self.data['hosts']}\n\n\n")
return list(self.data["hosts"].keys())
return self.data["globals"].get(key)
class FileHandler: class FileHandler:
CONFIG_FILE = "pywebsrv.conf" CONFIG_FILE = "pywebsrv.conf"
new_conf = "new_conf.conf" new_conf = "new_conf.conf"
def __init__(self, base_dir=None): def __init__(self, base_dir=None):
# this is a fucking clusterfuck.
self.config_path = os.path.join(os.getcwd(), self.CONFIG_FILE) self.config_path = os.path.join(os.getcwd(), self.CONFIG_FILE)
self.new_conf = os.path.join(os.getcwd(), self.new_conf) self.new_conf = os.path.join(os.getcwd(), self.new_conf)
self.base_dir = self.read_config("directory") self.base_dir = self.read_config("directory")
self.cached_conf = None with open(self.new_conf, "r") as f:
self.cfg = ConfigParser(f.read())
if not os.path.exists(self.config_path): if not os.path.exists(self.config_path):
print( print(
"The pywebsrv.conf file needs to be in the same directory " "The pywebsrv.conf file needs to be in the same directory "
@@ -75,11 +131,15 @@ class FileHandler:
"https://git.novacow.ch/Nova/PyWebServer/raw/branch/main/pywebsrv.conf" "https://git.novacow.ch/Nova/PyWebServer/raw/branch/main/pywebsrv.conf"
) )
exit(1) exit(1)
# TODO: fix this please!!
def read_file(self, file_path): def read_file(self, file_path, directory=None):
if "../" in file_path: if "../" in file_path or "%" in file_path:
return 403, None return 403, None
if directory is not None:
full_path = os.path.join(directory, file_path.lstrip("/"))
else:
full_path = os.path.join(self.base_dir, file_path.lstrip("/")) full_path = os.path.join(self.base_dir, file_path.lstrip("/"))
if not os.path.isfile(full_path): if not os.path.isfile(full_path):
return 404, None return 404, None
@@ -92,8 +152,8 @@ class FileHandler:
print(f"Error reading file {full_path}: {e}") print(f"Error reading file {full_path}: {e}")
return 500, None return 500, None
def write_file(self, file_path, data): def write_file(self, file_path, data, directory=None):
if "../" in file_path: if "../" in file_path or "%" in file_path:
return 403 return 403
full_path = os.path.join(self.base_dir, file_path.lstrip("/")) full_path = os.path.join(self.base_dir, file_path.lstrip("/"))
with open(full_path, "a") as f: with open(full_path, "a") as f:
@@ -117,7 +177,7 @@ class FileHandler:
"disable-autocertgen", "disable-autocertgen",
"key-file", "key-file",
"cert-file", "cert-file",
"block-ua" "block-ua",
] ]
if option not in valid_options: if option not in valid_options:
return None return None
@@ -165,57 +225,19 @@ class FileHandler:
return value return value
return None return None
def read_new_config(self, option, host=None): def read_new_config(self, key, host_name=None):
""" print(
Reads the configuration file and returns a dict f"\n\n\nQuery!\nkey: {key}\nhost_name: {host_name}\nret: {self.cfg.query_config(key, host_name)}"
"""
if self.cached_conf is None:
with open(self.new_conf, "r", encoding="utf-8") as fh:
text = fh.read()
blocks = re.findall(
r'^(host\s+(\S+)|globals)\s*\{([^}]*)\}', text, re.MULTILINE
) )
parsed = {} return self.cfg.query_config(key, host_name)
host_list = []
print(f"Blocks: {blocks}")
for tag, hostname, body in blocks:
section = hostname if hostname else "globals"
if hostname:
host_list.append(hostname)
kv = {}
for line in body.splitlines():
line = line.strip()
if not line or ":" not in line or line.startswith("#"):
continue
key, rest = line.split(":", 1)
key = key.strip()
rest = rest.strip()
# Split comma-separated values (e.g. GET,PUT)
if "," in rest:
kv[key] = [item.strip() for item in rest.split(",")]
else:
kv[key] = rest
parsed[section] = kv
parsed["globals"]["hosts"] = host_list
self.cached_conf = parsed
else:
parsed = self.cached_conf
if option == "host":
try:
return host_list
except Exception:
return parsed["globals"]["hosts"]
section = parsed.get(host or "globals", {})
return section.get(option)
def autocert(self): def autocert(self):
""" """
Generate some self-signed certificates using AutoCertGen Generate some self-signed certificates using AutoCertGen
TODO: doesn't work, need to fix. probably add `./` to $PATH TODO: doesn't work, need to fix. probably add `./` to $PATH
""" """
if not os.getcwd() in sys.path:
sys.path.append(f"{os.getcwd()}")
autocert = AutoCertGen() autocert = AutoCertGen()
autocert.gen_cert() autocert.gen_cert()
@@ -223,7 +245,7 @@ class FileHandler:
class RequestParser: class RequestParser:
def __init__(self): def __init__(self):
self.file_handler = FileHandler() self.file_handler = FileHandler()
self.hosts = self.file_handler.read_new_config("host") self.hosts = self.file_handler.read_new_config("hosts")
print(f"Hosts: {self.hosts}") print(f"Hosts: {self.hosts}")
def parse_request_line(self, line): def parse_request_line(self, line):
@@ -267,20 +289,23 @@ class RequestParser:
Mfw im in an ugly code writing contest and my opponent is nova while writing a side project Mfw im in an ugly code writing contest and my opponent is nova while writing a side project
""" """
host = f"{host}" host = f"{host}"
print(f"hosts: {self.hosts}, host: {host}") print(f"hosts: {self.hosts}, host: {host}, split: {host.rsplit(':', 1)[0]}")
if ":" in host: if ":" in host:
host = host.split(":", 1)[0] host = host.rsplit(":", 1)[0]
host = host.lstrip() host = host.lstrip()
host = host.rstrip() host = host.rstrip()
if ( if (
host == "localhost" or host == "127.0.0.1" host == "localhost" or host == "127.0.0.1" or host == "[::1]"
) and self.file_handler.read_config("allow-localhost"): ) and self.file_handler.read_new_config("allow-localhost"):
return True
if self.hosts is None:
return True return True
if host not in self.hosts: if host not in self.hosts:
return False return False
else: else:
return True return True
# #
# class ProxyServer: # class ProxyServer:
# def __init__( # def __init__(
@@ -294,10 +319,10 @@ class WebServer:
): ):
self.http_port = int(http_port) self.http_port = int(http_port)
self.https_port = int(https_port) self.https_port = int(https_port)
self.cert_file = cert_file
self.key_file = key_file
self.file_handler = FileHandler() self.file_handler = FileHandler()
self.parser = RequestParser() self.parser = RequestParser()
self.cert_file = self.file_handler.read_new_config("cert") or cert_file
self.key_file = self.file_handler.read_new_config("key") or key_file
self.skip_ssl = False self.skip_ssl = False
# me when no certificate and key file # me when no certificate and key file
@@ -324,14 +349,22 @@ class WebServer:
"This host cannot be reached without sending a `Host` header." "This host cannot be reached without sending a `Host` header."
) )
# ipv6 when????/??//?????//? # TODO: enable experimental ipv6 support in config
self.http_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.http_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.http_socket.bind(("0.0.0.0", self.http_port))
self.https_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # ipv6 when????/??//?????//?
self.https_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) # self.http_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.https_socket.bind(("0.0.0.0", self.https_port)) # self.http_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
# self.http_socket.bind(("0.0.0.0", self.http_port))
#
# self.https_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# self.https_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
# self.https_socket.bind(("0.0.0.0", self.https_port))
self.http_socket = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
self.http_socket.bind(("::", self.http_port))
self.https_socket = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
self.https_socket.bind(("::", self.https_port))
if self.skip_ssl is False: if self.skip_ssl is False:
# https gets the ssl treatment!! yaaaay :3 # https gets the ssl treatment!! yaaaay :3
@@ -367,17 +400,25 @@ class WebServer:
http_thread = threading.Thread(target=self.start_http, daemon=True) http_thread = threading.Thread(target=self.start_http, daemon=True)
https_thread = threading.Thread(target=self.start_https, daemon=True) https_thread = threading.Thread(target=self.start_https, daemon=True)
# ipv6http_thread = threading.Thread(target=self.start_http_ipv6, daemon=True)
# ipv6https_thread = threading.Thread(target=self.start_https_ipv6, daemon=True)
if https is True: if https is True:
if self.skip_ssl is True: if self.skip_ssl is True:
print("WARN: You have enabled HTTPS without SSL!!") print("WARN: You have enabled HTTPS without SSL!!")
yn = input("Is this intended behaviour? [y/N] ") yn = input("Is this intended behaviour? [y/N] ")
if yn.lower() == "n":
exit(1)
https_thread.start() https_thread.start()
# ipv6https_thread.start()
if http is True: if http is True:
# ipv6http_thread.start()
http_thread.start() http_thread.start()
http_thread.join() http_thread.join()
https_thread.join() https_thread.join()
# ipv6http_thread.join()
# ipv6https_thread.join()
def start_http(self): def start_http(self):
self.http_socket.listen(5) self.http_socket.listen(5)
@@ -391,6 +432,32 @@ class WebServer:
except OSError: except OSError:
break break
def start_http_ipv6(self):
self.ipv6http_socket.listen(5)
print(f"IPv6 HTTP server listening on port {self.http_port}...")
while self.running:
try:
conn, addr = self.ipv6http_socket.accept()
self.handle_connection(conn, addr)
except Exception as e:
print(f"HTTP error: {e}")
except OSError:
break
def start_https_ipv6(self):
self.ipv6https_socket.listen(5)
print(f"IPv6 HTTPS server listening on port {self.https_port}...")
while self.running:
try:
conn, addr = self.ipv6https_socket.accept()
self.handle_connection(conn, addr)
except Exception as e:
print(
f"HTTPS error: {e}"
) # be ready for ssl errors if you use a self-sign!!
except OSError:
break
def start_https(self): def start_https(self):
self.https_socket.listen(5) self.https_socket.listen(5)
print(f"HTTPS server listening on port {self.https_port}...") print(f"HTTPS server listening on port {self.https_port}...")
@@ -410,7 +477,9 @@ class WebServer:
data = conn.recv(512) data = conn.recv(512)
request = data.decode(errors="ignore") request = data.decode(errors="ignore")
if not data: if not data:
response = self.build_response(400, "Bad Request") # user did fucky-wucky response = self.build_response(
400, "Bad Request"
) # user did fucky-wucky
elif len(data) > 8192: elif len(data) > 8192:
response = self.build_response(413, "Request too long") response = self.build_response(413, "Request too long")
else: else:
@@ -440,9 +509,7 @@ class WebServer:
) )
break break
else: else:
return self.build_response( return self.build_response(400, self.no_host_req_response.encode())
400, self.no_host_req_response.encode()
)
for line in data.splitlines(): for line in data.splitlines():
if "User-Agent" in line: if "User-Agent" in line:
@@ -454,9 +521,7 @@ class WebServer:
) )
break break
else: else:
return self.build_response( return self.build_response(400, "You cannot connect without a User-Agent.")
400, "You cannot connect without a User-Agent."
)
method, path, version = self.parser.parse_request_line(request_line) method, path, version = self.parser.parse_request_line(request_line)
@@ -470,12 +535,18 @@ class WebServer:
self.parser = RequestParser() self.parser = RequestParser()
return self.build_response(302, "") return self.build_response(302, "")
if not self.parser.is_method_allowed( if not self.parser.is_method_allowed(method):
method
):
return self.build_response(405, self.http_405_html) return self.build_response(405, self.http_405_html)
file_content, mimetype = self.file_handler.read_file(path) if ":" in host:
host2 = host.rsplit(":", 1)[0]
directory = (
self.file_handler.read_new_config("directory", host2)
or self.file_handler.base_dir
)
file_content, mimetype = self.file_handler.read_file(path, directory)
if file_content == 403: if file_content == 403:
print("WARN: Directory traversal attack prevented.") # look ma, security!! print("WARN: Directory traversal attack prevented.") # look ma, security!!
@@ -494,7 +565,9 @@ class WebServer:
mimetype = mimetype[0] mimetype = mimetype[0]
if mimetype is None: if mimetype is None:
# We have to assume it's binary. # We have to assume it's binary.
return self.build_binary_response(200, file_content, "application/octet-stream") return self.build_binary_response(
200, file_content, "application/octet-stream"
)
if "text/" not in mimetype: if "text/" not in mimetype:
return self.build_binary_response(200, file_content, mimetype) return self.build_binary_response(200, file_content, mimetype)
@@ -508,7 +581,7 @@ class WebServer:
403: "Forbidden", 403: "Forbidden",
404: "Not Found", 404: "Not Found",
405: "Method Not Allowed", 405: "Method Not Allowed",
500: "Internal Server Error" 500: "Internal Server Error",
} }
status_message = messages.get(status_code) status_message = messages.get(status_code)
headers = ( headers = (
@@ -539,7 +612,7 @@ class WebServer:
405: "Method Not Allowed", 405: "Method Not Allowed",
413: "Payload Too Large", 413: "Payload Too Large",
500: "Internal Server Error", 500: "Internal Server Error",
635: "Go Away" 635: "Go Away",
} }
status_message = messages.get(status_code) status_message = messages.get(status_code)
@@ -560,7 +633,9 @@ class WebServer:
# Why not 307, Moved Permanently? Because browsers will cache the # Why not 307, Moved Permanently? Because browsers will cache the
# response and not send the reload command. # response and not send the reload command.
host = self.file_handler.read_config("host")[0] host = self.file_handler.read_config("host")[0]
port = self.file_handler.read_config("port-https") or self.file_handler.read_config("port") port = self.file_handler.read_config(
"port-https"
) or self.file_handler.read_config("port")
if port != 80 and port != 443: if port != 80 and port != 443:
if port == 8443: if port == 8443:
host = f"https://{host}:{port}/" host = f"https://{host}:{port}/"