6 Commits

Author SHA1 Message Date
Nova b48b682da7 html, branding change and getting ready for pre-rel 2026-03-24 22:50:26 +01:00
Nova 97896c87a7 idk anymore man 2026-03-13 22:13:59 +01:00
Nova 127612d408 smartass 2026-03-03 22:27:56 +01:00
Nova 3ff7a33695 something pog 2026-03-03 22:26:19 +01:00
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
6 changed files with 266 additions and 160 deletions
+23
View File
@@ -6,4 +6,27 @@ promised features missing, but I'm very much working on it live!
Every save I do increments the build number by 1, I won't publish all of them, but most of them will be published.
Once a milestone is hit (e.g. a new feature fully implemented), I'll publish a release!
## Currently working features:
* New configuration is ~75% done, most features work.
* Fixed **A LOT** of unreported bugs from the old code.
* More resilliency against errors.
* Improved security.
## Project status:
Amethyst will stay in beta for a while, I want all features to work, put I will make pre-release versions that are mostly stable.
They can be found as the `amethyst-prerel-0.a.b` releases. I won't guarantee 100% stability, but waay more than just some random build.
## Install instructions:
Install Python, and change the provided config.
## Minimum requirements:
Python 3.8+
And whatever PC that happens to run that.
I recommend Python 3.10 or above though, with a PC running:
* Windows 8.1+
* macOS 10.15+
* Linux 4.19+
* FreeBSD 13.2R+
* Some other somewhat recent OS.
## Currently W.I.P. Check back later!
+26
View File
@@ -0,0 +1,26 @@
"""
This is the Amethyst API mode Python interface whatevers.
Docs will follow.
"""
# Below go imports.
import sys
import os
if not os.getcwd() in sys.path:
sys.path.append(os.getcwd())
import pywebsrv
class API:
"""
class
"""
def __init__(self):
# DO NOT USE THIS CLASS FOR PROGRAM, ONLY ON_REQUEST PLEASE!!
# Below go definitions to get things working.
self.build_response = pywebsrv.WebServer.build_binary_response
def on_request(self, req):
return self.build_response(200, "This is a test", "text/html")
+6 -5
View File
@@ -4,10 +4,11 @@
<title>Test page</title>
</head>
<body>
<h1>Hey there!</h1>
<h2>You're seeing this page because you haven't set up PyWebServer yet!</h2>
<h2>This page confirms that PyWebServer can read and serve files from your PC.</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>Here you can simulate a 404 error: <a href="/uuh">Click me for a 404 error!</a></p>
<center>
<h1>Hello from Amethyst!</h1>
<h2>This page confirms Amethyst can read files from your PC or server and serve them to your browser!</h2>
<p>This is a test page, if you aren't the server owner, they might not have finished setting up their site, be patient. If this doesn't go away after a while, tell them they've made an oopsie</p>
<p>This server runs Amethyst Pre-Rel Build 0053</p>
</center>
</body>
</html>
+14 -12
View File
@@ -1,23 +1,24 @@
# WARNING: This is an alpha spec of NSCL 2.0!!
host example.com {
host 192.168.2.196 {
directory:/home/nova/Downloads/test/html
allowed-methods:GET
block-ip:match-ip("192.168",2)
block-ua:match("Discordbot",0),match("Google",0)
block-ua:match("Discordbot"),match("Google")
}
host cdn.example.com {
directory:/home/nova/Downloads/test/cdn
host localhost {
directory:/home/nova/PyWebServer/html2
allowed-methods:GET,PUT
block-ip:10.1.100.2
block-ua:match("Discordbot",0)
apimode:0
block-ua:match("Discordbot")
}
host modem.example.com {
proxy:192.168.2.254
key-file:/home/nova/Downloads/test/proxykey.pem
cert-file:/home/nova/Downloads/test/proxycert.pem
host 192.168.1.213 {
directory:/home/nova/PyWebServer/html
allowed-methods:GET,PUT
block-ip:10.1.100.2
block-ua:match("Discordbot")
}
globals {
@@ -25,6 +26,7 @@ globals {
https:1
port:8080
https-port:8443
global-key:/home/nova/Downloads/test/key.pem
global-cert:/home/nova/Downloads/test/cert.pem
allow-localhost:1
key:/home/nova/PyWebServer/ssl/key.pem
cert:/home/nova/PyWebServer/ssl/cert.pem
}
+4 -4
View File
@@ -4,11 +4,11 @@
port:8080
port-https:8443
# 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.
# You can use FQDNs, IP-addresses and localhost,
# Support for multiple hosts is coming.
host:localhost
host:localhost,10.185.213.118
# Enables HTTP support. (Only enables/disables the HTTP port.)
http:1
# Enables HTTPS support. (Only enables/disables the HTTPS port.)
@@ -21,8 +21,8 @@ allow-localhost:1
disable-autocertgen:0
# 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
# If you wish to block User-Agents, this function is coming though.
# block-ua:(NULL)
# If you wish to block User-Agents.
block-ua:match(Discordbot),match(google)
# TEST: experimental non-defined keys go here:
# keyfile key
+193 -139
View File
@@ -35,6 +35,8 @@ Library aswell as a standalone script:
You can easily get access to other parts of the script if you need it.
TODO: actually put normal comments in
TODO: INPROG: add typing to all code, new code will feature it by default.
"""
import os
@@ -42,32 +44,91 @@ import mimetypes
import threading
import ssl
import socket
import re
# import re
import signal
import sys
try:
if not os.getcwd() in sys.path:
sys.path.append(f"{os.getcwd()}")
from certgen import AutoCertGen
except ImportError:
# just do nothing, it's not working anyway.
# print(
# "WARN: You need the AutoCertGen plugin! Please install it from\n"
# "https://git.novacow.ch/Nova/AutoCertGen/"
# )
pass
print(
"WARN: You need the AutoCertGen plugin! Please install it from\n"
"https://git.novacow.ch/Nova/AutoCertGen/"
)
# pass
AMETHYST_BUILD_NUMBER = "0001"
AMETHYST_BUILD_NUMBER = "0053"
AMETHYST_REPO = "https://git.novacow.ch/Nova/PyWebServer/"
class ConfigParser:
def __init__(self, text):
self.data: dict = {"hosts": {}, "globals": {}}
self._parse(text)
def _parse(self, text):
lines: list = [
line.strip()
for line in text.splitlines()
if line.strip() and not line.strip().startswith("#")
]
current_block: tuple | None = None
current_name: str | None = 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: str = key.strip()
value: str = 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:
CONFIG_FILE = "pywebsrv.conf"
new_conf = "new_conf.conf"
def __init__(self, base_dir=None):
# this is a fucking clusterfuck.
self.config_path = os.path.join(os.getcwd(), self.CONFIG_FILE)
self.new_conf = os.path.join(os.getcwd(), self.new_conf)
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):
print(
"The pywebsrv.conf file needs to be in the same directory "
@@ -75,12 +136,18 @@ class FileHandler:
"https://git.novacow.ch/Nova/PyWebServer/raw/branch/main/pywebsrv.conf"
)
exit(1)
# TODO: fix this please!!
def read_file(self, file_path):
if "../" in file_path:
def read_file(self, file_path, directory=None):
if "../" in file_path or "%" in file_path:
return 403, None
if file_path == "api.py":
return 404, None
full_path = os.path.join(self.base_dir, file_path.lstrip("/"))
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("/"))
if not os.path.isfile(full_path):
return 404, None
@@ -92,8 +159,8 @@ class FileHandler:
print(f"Error reading file {full_path}: {e}")
return 500, None
def write_file(self, file_path, data):
if "../" in file_path:
def write_file(self, file_path, data, directory=None):
if "../" in file_path or "%" in file_path:
return 403
full_path = os.path.join(self.base_dir, file_path.lstrip("/"))
with open(full_path, "a") as f:
@@ -117,7 +184,7 @@ class FileHandler:
"disable-autocertgen",
"key-file",
"cert-file",
"block-ua"
"block-ua",
]
if option not in valid_options:
return None
@@ -142,7 +209,7 @@ class FileHandler:
if val.startswith("match(") and val.endswith(")"):
idx = val.index("(")
idx2 = val.index(")")
ua_to_match = val[idx+1:idx2]
ua_to_match = val[idx + 1 : idx2]
host_to_match.append(ua_to_match)
else:
literal_blocks.append(val)
@@ -165,51 +232,11 @@ class FileHandler:
return value
return None
def read_new_config(self, option, host=None):
"""
Reads the configuration file and returns a dict
"""
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 = {}
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 read_new_config(self, key, host_name=None):
print(
f"\n\n\nQuery!\nkey: {key}\nhost_name: {host_name}\nret: {self.cfg.query_config(key, host_name)}"
)
return self.cfg.query_config(key, host_name)
def autocert(self):
"""
@@ -223,42 +250,51 @@ class FileHandler:
class RequestParser:
def __init__(self):
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}")
def parse_request_line(self, line):
def parse_request_line(self, line, host):
"""Parses the HTTP request line."""
try:
method, path, version = line.split(" ")
except ValueError:
return None, None, None
if path.endswith("/"):
path += "index.html"
return "DELETE", "/this/is/a/bogus/request", "HTTP/1.0"
if path.endswith("/") or ("." not in path):
if not path.endswith("/"):
path += "/"
index = self.file_handler.read_new_config("index", host) or "index.html"
path += f"{index}"
return method, path, version
def ua_blocker(self, ua, host=None):
def ua_is_allowed(self, ua, host=None):
"""Parses and matches UA to block"""
del host
match, literal = self.file_handler.read_config("block-ua")
if ua in literal:
return False
for _ua in match:
if _ua.lower() in ua.lower():
return False
return True
# del host
# _list = self.file_handler.read_config("block-ua")
# if _list is None:
# return True
# match, literal = self.file_handler.parse_match_blocks(_list)
# if ua in literal:
# return False
# for _ua in match:
# if _ua.lower() in ua.lower():
# return False
# return True
def is_method_allowed(self, method):
def is_method_allowed(self, method, host=None):
"""
Checks if the HTTP method is allowed.
Reads allowed methods from a configuration file.
Falls back to allowing only 'GET' if the file does not exist.
Should (for now) only be GET as I haven't implemented the logic for PUT
"""
allowed_methods = ["GET"]
# allowed_methods = ["GET"]
# While the logic for PUT, DELETE, etc. is not added, we shouldn't
# allow for it to attempt it.
# Prepatched for new update.
# allowed_methods = self.file_handler.read_config("allowed-methods")
allowed_methods = self.file_handler.read_new_config("allowed-methods", host)
if allowed_methods is None:
allowed_methods = ["GET"]
return method in allowed_methods
def host_parser(self, host):
@@ -267,26 +303,22 @@ class RequestParser:
Mfw im in an ugly code writing contest and my opponent is nova while writing a side project
"""
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:
host = host.split(":", 1)[0]
host = host.rsplit(":", 1)[0]
host = host.lstrip()
host = host.rstrip()
if (
host == "localhost" or host == "127.0.0.1"
) and self.file_handler.read_config("allow-localhost"):
host == "localhost" or host == "127.0.0.1" or host == "[::1]"
) and self.file_handler.read_new_config("allow-localhost"):
return True
if self.hosts is None:
return True
if host not in self.hosts:
return False
else:
return True
#
# class ProxyServer:
# def __init__(
# self,
# ):
class WebServer:
def __init__(
@@ -294,10 +326,10 @@ class WebServer:
):
self.http_port = int(http_port)
self.https_port = int(https_port)
self.cert_file = cert_file
self.key_file = key_file
self.file_handler = FileHandler()
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
# me when no certificate and key file
@@ -324,14 +356,11 @@ class WebServer:
"This host cannot be reached without sending a `Host` header."
)
# ipv6 when????/??//?????//?
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.http_socket = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
self.http_socket.bind(("::", 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.https_socket = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
self.https_socket.bind(("::", self.https_port))
if self.skip_ssl is False:
# https gets the ssl treatment!! yaaaay :3
@@ -344,18 +373,18 @@ class WebServer:
)
self.http_404_html = (
"<html><head><title>HTTP 404 - PyWebServer</title></head>"
f"<body><center><h1>HTTP 404 - Not Found!</h1><p>Running PyWebServer/amethyst-build-{AMETHYST_BUILD_NUMBER}</p>"
"<html><head><title>HTTP 404 - Amethyst</title></head>"
f"<body><center><h1>HTTP 404 - Not Found!</h1><p>Running Amethyst/build-{AMETHYST_BUILD_NUMBER}</p>"
"</center></body></html>"
)
self.http_403_html = (
"<html><head><title>HTTP 403 - PyWebServer</title></head>"
f"<body><center><h1>HTTP 403 - Forbidden</h1><p>Running PyWebServer/amethyst-build-{AMETHYST_BUILD_NUMBER}</p>"
"<html><head><title>HTTP 403 - Amethyst</title></head>"
f"<body><center><h1>HTTP 403 - Forbidden</h1><p>Running Amethyst/build-{AMETHYST_BUILD_NUMBER}</p>"
"</center></body></html>"
)
self.http_405_html = (
"<html><head><title>HTTP 405 - PyWebServer</title></head>"
f"<body><center><h1>HTTP 405 - Method not allowed</h1><p>Running PyWebServer/amethyst-build-{AMETHYST_BUILD_NUMBER}</p>"
"<html><head><title>HTTP 405 - Amethyst</title></head>"
f"<body><center><h1>HTTP 405 - Method not allowed</h1><p>Running Amethyst/build-{AMETHYST_BUILD_NUMBER}</p>"
"</center></body></html>"
)
@@ -372,6 +401,8 @@ class WebServer:
if self.skip_ssl is True:
print("WARN: You have enabled HTTPS without SSL!!")
yn = input("Is this intended behaviour? [y/N] ")
if yn.lower() == "n":
exit(1)
https_thread.start()
if http is True:
http_thread.start()
@@ -410,7 +441,9 @@ class WebServer:
data = conn.recv(512)
request = data.decode(errors="ignore")
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:
response = self.build_response(413, "Request too long")
else:
@@ -422,6 +455,13 @@ class WebServer:
conn.sendall(response)
except Exception as e:
print(f"Error handling connection: {e}")
response = self.build_response(
500,
"Amethyst is currently unable to serve your request. Below is debug info.\r\n"
f"Error: {e}; Version: amethyst-b{AMETHYST_BUILD_NUMBER}\r\n"
"You cannot do anything at this time, the server owner has made a misconfiguration or there is a bug in the program",
)
conn.sendall(response)
finally:
conn.close()
@@ -440,25 +480,26 @@ class WebServer:
)
break
else:
return self.build_response(
400, self.no_host_req_response.encode()
)
return self.build_response(400, self.no_host_req_response.encode())
for line in data.splitlines():
if "User-Agent" in line:
ua = line.split(":", 1)[1].strip()
allowed = self.parser.ua_blocker(ua)
allowed = self.parser.ua_is_allowed(ua)
if not allowed:
return self.build_response(
403, "This UA has been blocked by the owner of this site."
)
break
else:
return self.build_response(
400, "You cannot connect without a User-Agent."
)
return self.build_response(400, "You cannot connect without a User-Agent.")
method, path, version = self.parser.parse_request_line(request_line)
if ":" in host:
host2 = host.rsplit(":", 1)[0]
else:
host2 = host
method, path, version = self.parser.parse_request_line(request_line, host2)
if not all([method, path, version]):
return self.build_response(400, "Bad Request")
@@ -468,14 +509,25 @@ class WebServer:
print("Got reload command! Reloading configuration...")
self.file_handler = FileHandler()
self.parser = RequestParser()
return self.build_response(302, "")
return self.build_response(302, "", host=host2)
if not self.parser.is_method_allowed(
method
):
if not self.parser.is_method_allowed(method):
return self.build_response(405, self.http_405_html)
file_content, mimetype = self.file_handler.read_file(path)
directory = (
self.file_handler.read_new_config("directory", host2)
or self.file_handler.base_dir
)
if self.file_handler.read_new_config("apimode", host2) is True:
if not os.path.join(os.getcwd(), directory) in sys.path:
sys.path.append(f"{os.path.join(os.getcwd(), directory)}")
import api
apiclass = api.API()
return apiclass.on_request(data)
file_content, mimetype = self.file_handler.read_file(path, directory)
if file_content == 403:
print("WARN: Directory traversal attack prevented.") # look ma, security!!
@@ -485,16 +537,16 @@ class WebServer:
if file_content == 500:
return self.build_response(
500,
"PyWebServer has encountered a fatal error and cannot serve "
"Amethyst has encountered a fatal error and cannot serve "
"your request. Contact the owner with this error: FATAL_FILE_RO_ACCESS",
) # When there was an issue with reading we throw this.
# A really crude implementation of binary files. Later in 2.0 I'll actually
# make this useful.
mimetype = mimetype[0]
if mimetype is None:
# 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:
return self.build_binary_response(200, file_content, mimetype)
@@ -508,22 +560,22 @@ class WebServer:
403: "Forbidden",
404: "Not Found",
405: "Method Not Allowed",
500: "Internal Server Error"
500: "Internal Server Error",
}
status_message = messages.get(status_code)
headers = (
f"HTTP/1.1 {status_code} {status_message}\r\n"
f"Server: PyWebServer/amethyst-build-{AMETHYST_BUILD_NUMBER}\r\n"
f"Server: Amethyst/amethyst-build-{AMETHYST_BUILD_NUMBER}\r\n"
f"Content-Type: {content_type}\r\n"
f"Content-Length: {len(binary_data)}\r\n"
f"Connection: close\r\n\r\n"
# Connection close is done because it is way easier to implement.
# It's not like this program will see production use anyway.
# Tbh when i'll implement HTTP2
)
return headers.encode() + binary_data
def build_response(self, status_code, body):
@staticmethod
def build_response(status_code, body, host=None):
"""
For textfiles we'll not have to guess MIME-types, though the other function
build_binary_response will be merged in here anyway.
@@ -539,7 +591,7 @@ class WebServer:
405: "Method Not Allowed",
413: "Payload Too Large",
500: "Internal Server Error",
635: "Go Away"
621: "fuck off! :3",
}
status_message = messages.get(status_code)
@@ -550,7 +602,7 @@ class WebServer:
# Don't encode yet, if 302 status code we have to include location.
headers = (
f"HTTP/1.1 {status_code} {status_message}\r\n"
f"Server: PyWebServer/amethyst-build-{AMETHYST_BUILD_NUMBER}\r\n"
f"Server: Amethyst/build-{AMETHYST_BUILD_NUMBER}\r\n"
f"Content-Length: {len(body)}\r\n"
f"Connection: close\r\n\r\n"
).encode()
@@ -559,26 +611,28 @@ class WebServer:
# 302 currently only happens when the reload is triggered.
# Why not 307, Moved Permanently? Because browsers will cache the
# response and not send the reload command.
host = self.file_handler.read_config("host")[0]
port = self.file_handler.read_config("port-https") or self.file_handler.read_config("port")
if port != 80 and port != 443:
if port == 8443:
host = f"https://{host}:{port}/"
else:
host = f"http://{host}:{port}/"
else:
if port == 443:
host = f"https://{host}/"
else:
host = f"http://{host}/"
# if port == 443:
# host = f"https://{host}/"
# else:
# host = f"http://{host}/"
headers = (
f"HTTP/1.1 {status_code} {status_message}\r\n"
f"Location: {host}\r\n"
f"Server: PyWebServer/amethyst-build-{AMETHYST_BUILD_NUMBER}\r\n"
f"Server: Amethyst/build-{AMETHYST_BUILD_NUMBER}\r\n"
f"Content-Length: {len(body)}\r\n"
f"Connection: close\r\n\r\n"
).encode()
if status_code == 621:
headers = (
f"HTTP/1.1 {status_code} {status_message}\r\n"
"Server: Amethyst/build-0621\r\n"
"Content-Length: 30\r\n"
f"Connection: close\r\n\r\n"
)
body = "https://e621.net/posts/6155664"
print(f"{headers + body}")
return headers + body
def shutdown(self, signum, frame):
@@ -604,7 +658,7 @@ def main():
file_handler = FileHandler()
file_handler.base_dir = file_handler.read_config("directory")
http_port = file_handler.read_new_config("port") or 8080
https_port = file_handler.read_new_config("port-https") or 8443
https_port = file_handler.read_new_config("https-port") or 8443
http_enabled = bool(file_handler.read_new_config("http")) or True
print(http_enabled)
https_enabled = bool(file_handler.read_new_config("https")) or False