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

Compare commits

base: Nova:2.0
Branches Tags
Nova:main Nova:2.0 Nova:backport-1.0-issue4
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:1.0+u2
Branches Tags
Nova:2.0 Nova:main Nova:backport-1.0-issue4
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
2.0 ... 1.0+u2

Author SHA1 Message Date
Nova
0e406ab360 Quick patch to get certgen running 2025-05-04 14:57:58 +02:00
Nova
f29ad0d6b4 Backport urgent fix to 1.0 2025-05-04 00:03:08 +02:00
5 changed files with 124 additions and 437 deletions
Expand all files Collapse all files

9
README.md
View File

@@ -1,9 +1,2 @@
# Amethyst Web Server
# PyWebServer
## A word of warning!
Currently Amethyst is in very early alpha stage, a lot of things will be broken, names won't be correct,
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 W.I.P. Check back later!

21
html/index.html
View File

@@ -1,21 +0,0 @@
<!DOCTYPE html>
<html>
<head>
<title>Test page</title>
<link rel="stylesheet" href="/style.css">
</head>
<body>
<center>
<h1>Web portal</h1>
<h1><a href="/d.mp4">d.mp4</a></h1>
<h1><a href="/webjammies/index.ini">index.ini</a></h1>
<h1><a href="/webjammies/rabado.mp3">rabado.mp3</a></h1>
<h1><a href="/webjammies/invaders.mp3">invaders.mp3</a></h1>
<h1><a href="/webjammies/momentum.mp3">momentum.mp3</a></h1>
<h1><a href="/webjammies/bacterial-love.mp3">bacterial-love.mp3</a></h1>
<h1><a href="/webjammies/old-money-bitch.mp3">old-money-bitch.mp3</a></h1>
<h1><a href="/webjammies/ravestate909.mp3">ravestate909.mp3</a></h1>
<h1><a href="/webjammies/leash.mp3">leash.mp3</a></h1>
</center>
</body>
</html>

36
new_conf.conf
View File

@@ -1,36 +0,0 @@
# WARNING: This is an alpha spec of NSCL 2.0!!
host 192.168.1.196 {
location /secrets {
root:/var/www/html/hidden
whitelist-ua:"Mozilla/5.0 (X11; Linux x86_64) Gecko/20100101 Firefox/149.0 Furfox/1.0"
whitelist-ip:123.45.67.89
}
directory:/home/nova/Downloads/test/html
allowed-methods:GET
block-ip:match-ip("192.168",2)
block-ua:match("Discordbot",0),match("Google",0)
}
host cdn.example.com {
directory:/home/nova/Downloads/test/cdn
allowed-methods:GET,PUT
block-ip:10.1.100.2
block-ua:match("Discordbot",0)
}
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
}
globals {
http:1
https:1
port:8080
https-port:8443
allow-localhost:1
global-key:/home/nova/Downloads/test/key.pem
global-cert:/home/nova/Downloads/test/cert.pem
}

38
pywebsrv.conf
View File

@@ -1,33 +1,17 @@
# Using NSCL 1.3
# Port defenition. What ports to use.
# port is the HTTP port, port-https is the HTTPS port
port:8080
directory:/home/nova/Documents/html
host:localhost
# DANGER: NEVER EVER TURN THIS ON IN PROD!!!!!!!!!!!!
allow-all:1
# DANGER!!
port-https:8443
# Here you choose what directory PyWebServer looks in for files.
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,10.185.213.118
# Enables HTTP support. (Only enables/disables the HTTP port.)
http:1
# Enables HTTPS support. (Only enables/disables the HTTPS port.)
https:1
# Allows the use of localhost to connect.
# The default is on, this is seperate of the host defenition.
allow-localhost:1
# If you're using the webserver in a library form,
# you can disable the AutoCertGen and never trigger it.
disable-autocertgen:0
# If you wish to block IP-addresses, this function is coming though.
allow-localhost:0
# for use in libraries
# disable-autocertgen:0
# block-ip:0.0.0.0,1.1.1.1,2.2.2.2
# If you wish to block User-Agents.
block-ua:match(Discordbot),match(google)
# TEST: experimental non-defined keys go here:
# keyfile key
key-file:/home/nova/PyWebServer/key.pem
# certfile keys
cert-file:/home/nova/PyWebServer/cert.pem
# allowed-methods, csv's
allowed-methods:GET
# block-ua:(NULL)
# allow-nohost:0
# In libraries you can disable everything you don't need.

457
pywebsrv.py
View File

@@ -1,19 +1,4 @@
"""
License:
PyWebServer
Copyright (C) 2025 Nova
This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version.
This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.
You should have received a copy of the GNU General Public License along with this program. If not, see <https://www.gnu.org/licenses/>.
Contact:
E-mail: nova@novacow.ch
NOTE: Once 2.0 is released, PyWebServer will become the Amethyst Web Server
This is PyWebServer, an ultra minimalist webserver, meant to still have
a lot standard webserver features. A comprehensive list is below:
Features:
@@ -33,71 +18,63 @@ Simple to understand and mod codebase.
All GNU GPL-3-or-above license. (Do with it what you want.)
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
"""
import os
import mimetypes
import threading
import ssl
import socket
import re
import signal
import sys
try:
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
AMETHYST_BUILD_NUMBER = "0018"
AMETHYST_REPO = "https://git.novacow.ch/Nova/PyWebServer/"
print(
"WARN: You need the AutoCertGen plugin! Please install it from\n"
"https://git.novacow.ch/Nova/AutoCertGen/"
)
class FileHandler:
CONFIG_FILE = "pywebsrv.conf"
new_conf = "new_conf.conf"
DEFAULT_CONFIG = (
"port:8080\nport-https:8443\nhttp:1"
"\nhttps:0\ndirectory:{cwd}\nhost:localhost"
"allow-all:1\nallow-localhost:1"
)
def __init__(self, base_dir=None):
self.base_dir = base_dir or os.path.join(os.getcwd(), "html")
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
if not os.path.exists(self.config_path):
print(
"The pywebsrv.conf file needs to be in the same directory "
"as pywebsrv.py! Get the default config file from:\n"
"https://git.novacow.ch/Nova/PyWebServer/raw/branch/main/pywebsrv.conf"
)
exit(1)
def read_file(self, file_path, directory=None):
if "../" in file_path or "%" in file_path:
return 403, None
def check_first_run(self):
if not os.path.isfile(self.config_path):
self.on_first_run()
return True
return False
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("/"))
def on_first_run(self):
with open(self.config_path, "w") as f:
f.write(self.DEFAULT_CONFIG.format(cwd=os.getcwd()))
def read_file(self, file_path):
if "../" in file_path:
return 403
full_path = os.path.join(self.base_dir, file_path.lstrip("/"))
if not os.path.isfile(full_path):
return 404, None
return 404
try:
mimetype = mimetypes.guess_type(full_path)
with open(full_path, "rb") as f:
return f.read(), mimetype
return f.read()
except Exception as e:
print(f"Error reading file {full_path}: {e}")
return 500, None
return 500
def write_file(self, file_path, data, directory=None):
if "../" in file_path or "%" in file_path:
def write_file(self, file_path, data):
if "../" in file_path:
return 403
full_path = os.path.join(self.base_dir, file_path.lstrip("/"))
with open(full_path, "a") as f:
@@ -117,11 +94,9 @@ class FileHandler:
"http",
"https",
"port-https",
"allow-all",
"allow-localhost",
"disable-autocertgen",
"key-file",
"cert-file",
"block-ua"
]
if option not in valid_options:
return None
@@ -136,121 +111,45 @@ class FileHandler:
key = key.lower()
if key == option:
if option == "host":
seperated_values = value.split(",", -1)
seperated_values = value.split(",", 0)
return [value.lower() for value in seperated_values]
if option == "block-ua":
seperated_values = value.split(",", -1)
host_to_match = []
literal_blocks = []
for val in seperated_values:
if val.startswith("match(") and val.endswith(")"):
idx = val.index("(")
idx2 = val.index(")")
ua_to_match = val[idx+1:idx2]
host_to_match.append(ua_to_match)
else:
literal_blocks.append(val)
return host_to_match, literal_blocks
if option == "port" or option == "port-https":
return int(value)
if (
option == "http"
or option == "https"
or option == "allow-all"
or option == "allow-localhost"
or option == "disable-autocertgen"
):
return bool(int(value))
if option == "directory":
if value == "<Enter directory here>":
return os.path.join(os.getcwd(), "html")
if value.endswith("/"):
value = value.rstrip("/")
return value
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()
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):
"""
Generate some self-signed certificates using AutoCertGen
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.gen_cert()
class RequestParser:
def __init__(self):
self.allowed_methods_file = "allowedmethods.conf"
self.file_handler = FileHandler()
self.hosts = self.file_handler.read_new_config("host")
print(f"Hosts: {self.hosts}")
self.hosts = self.file_handler.read_config("host")
self.all_allowed = self.file_handler.read_config("allow-all")
def parse_request_line(self, line):
"""Parses the HTTP request line."""
try:
method, path, version = line.split(" ")
if path.endswith("/"):
path += "index.html"
return method, path, version
except ValueError:
return None, None, None
if path.endswith("/"):
path += "index.html"
return method, path, version
def ua_blocker(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
def is_method_allowed(self, method):
"""
@@ -260,10 +159,9 @@ class RequestParser:
Should (for now) only be GET as I haven't implemented the logic for PUT
"""
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")
if os.path.isfile(self.allowed_methods_file):
with open(self.allowed_methods_file, "r") as f:
allowed_methods = [line.strip() for line in f]
return method in allowed_methods
def host_parser(self, host):
@@ -271,34 +169,26 @@ class RequestParser:
Parses the host and makes sure it's allowed in
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}, split: {host.rsplit(":", 1)[0]}")
host = str(host)
if ":" in host:
host = host.rsplit(":", 1)[0]
host = host.split(":", 1)[0]
host = host.lstrip()
host = host.rstrip()
if (
host == "localhost" or host == "127.0.0.1" or host == "[::1]"
) and self.file_handler.read_new_config("allow-localhost"):
host == "localhost" or host == "127.0.0.1"
) and self.file_handler.read_config("allow-localhost"):
return True
if host not in self.hosts:
if host not in self.hosts and self.all_allowed is False:
return False
else:
elif host not in self.hosts and self.all_allowed is True:
return True
#
# class ProxyServer:
# def __init__(
# self,
# ):
class WebServer:
def __init__(
self, http_port=8080, https_port=8443, cert_file="cert.pem", key_file="key.pem"
):
self.http_port = int(http_port)
self.https_port = int(https_port)
self.http_port = http_port
self.https_port = https_port
self.cert_file = cert_file
self.key_file = key_file
self.file_handler = FileHandler()
@@ -307,13 +197,6 @@ class WebServer:
# me when no certificate and key file
if not os.path.exists(self.cert_file) or not os.path.exists(self.key_file):
if not os.path.exists(self.cert_file) and not os.path.exists(self.key_file):
pass
# maybe warn users we purge their key/cert files? xdd
elif not os.path.exists(self.cert_file):
os.remove(self.key_file)
elif not os.path.exists(self.key_file):
os.remove(self.cert_file)
print("WARN: No HTTPS certificate was found!")
if self.file_handler.read_config("disable-autocertgen") is True:
print("WARN: AutoCertGen is disabled, ignoring...")
@@ -325,26 +208,21 @@ class WebServer:
else:
self.skip_ssl = True
# TODO: change this to something like oh no you fucked up, go fix idiot
self.no_host_req_response = (
"This host cannot be reached without sending a `Host` header."
"Connecting via this host is disallowed\r\n"
"You may also be using a very old browser!\r\n"
"Ask the owner of this website to set allow-all to 1!"
)
# TODO: enable experimental ipv6 support in config
# 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.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_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_INET6, socket.SOCK_STREAM)
self.https_socket.bind(("::", self.https_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))
if self.skip_ssl is False:
# https gets the ssl treatment!! yaaaay :3
@@ -356,22 +234,6 @@ class WebServer:
self.https_socket, server_side=True
)
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>"
"</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>"
"</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>"
"</center></body></html>"
)
self.running = True
def start(self, http, https):
@@ -380,23 +242,18 @@ class WebServer:
http_thread = threading.Thread(target=self.start_http, 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 self.skip_ssl is True:
print("WARN: You have enabled HTTPS without SSL!!")
yn = input("Is this intended behaviour? [y/N] ")
https_thread.start()
# ipv6https_thread.start()
if http is True:
# ipv6http_thread.start()
http_thread.start()
print(
f"Server running:\n - HTTP on port {self.http_port}\n - HTTPS on port {self.https_port}"
)
http_thread.join()
https_thread.join()
# ipv6http_thread.join()
# ipv6https_thread.join()
def start_http(self):
self.http_socket.listen(5)
@@ -404,44 +261,20 @@ class WebServer:
while self.running:
try:
conn, addr = self.http_socket.accept()
print(f"HTTP connection received from {addr}")
self.handle_connection(conn, addr)
except Exception as e:
print(f"HTTP error: {e}")
except OSError:
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):
self.https_socket.listen(5)
print(f"HTTPS server listening on port {self.https_port}...")
while self.running:
try:
conn, addr = self.https_socket.accept()
print(f"HTTPS connection received from {addr}")
self.handle_connection(conn, addr)
except Exception as e:
print(
@@ -452,17 +285,12 @@ class WebServer:
def handle_connection(self, conn, addr):
try:
data = conn.recv(512)
data = conn.recv(512) # why? well internet and tutiorials
request = data.decode(errors="ignore")
if not data:
response = self.build_response(400, "Bad Request") # user did fucky-wucky
elif len(data) > 8192:
response = self.build_response(413, "Request too long")
else:
response = self.handle_request(request, addr)
response = self.handle_request(request, addr)
if isinstance(response, str):
response = response.encode()
response = response.encode() # if we send text this shouldn't explode
conn.sendall(response)
except Exception as e:
@@ -471,7 +299,9 @@ class WebServer:
conn.close()
def handle_request(self, data, addr):
print(f"data: {data}")
if not data:
return self.build_response(400, "Bad Request") # user did fucky-wucky
request_line = data.splitlines()[0]
# Extract host from headers, never works though
@@ -485,151 +315,101 @@ class WebServer:
)
break
else:
if (
self.file_handler.read_config("allow-nohost") is True
): # no host is stupid
pass
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)
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."
)
403, self.no_host_req_response.encode()
) # the default (i hope to god)
method, path, version = self.parser.parse_request_line(request_line)
if not all([method, path, version]):
return self.build_response(400, "Bad Request")
# Figure out a better way to reload config
if path == "/?pywebsrv_reload_conf=1":
print("Got reload command! Reloading configuration...")
self.file_handler = FileHandler()
self.parser = RequestParser()
return self.build_response(302, "")
if not self.parser.is_method_allowed(
if not all([method, path, version]) or not self.parser.is_method_allowed(
method
):
return self.build_response(405, self.http_405_html)
return self.build_response(405, "Method Not Allowed")
directory = self.file_handler.read_new_config("directory", host)
file_content, mimetype = self.file_handler.read_file(path, directory)
file_content = self.file_handler.read_file(path)
if file_content == 403:
print("WARN: Directory traversal attack prevented.") # look ma, security!!
return self.build_response(403, self.http_403_html)
return self.build_response(403, "Forbidden")
if file_content == 404:
return self.build_response(404, self.http_404_html)
return self.build_response(404, "Not Found")
if file_content == 500:
return self.build_response(
500,
"PyWebServer 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.
) # The user did no fucky-wucky, but the server fucking exploded.
# 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")
if "text/" not in mimetype:
return self.build_binary_response(200, file_content, mimetype)
# (try to) detect binary files (eg, mp3) and serve them correctly
if path.endswith((".mp3", ".png", ".jpg", ".jpeg", ".gif")):
return self.build_binary_response(200, file_content, path)
return self.build_response(200, file_content)
@staticmethod
def build_binary_response(status_code, binary_data, content_type):
def build_binary_response(status_code, binary_data, filename):
"""Handles binary files like MP3s."""
messages = {
200: "OK",
403: "Forbidden",
404: "Not Found",
405: "Method Not Allowed",
500: "Internal Server Error"
500: "Internal Server Error",
}
status_message = messages.get(status_code)
# In the spirit of keeping stuff small, we'll just guess and see.
content_type = "application/octet-stream"
if filename.endswith(".mp3"):
content_type = "audio/mpeg"
elif filename.endswith(".png"):
content_type = "image/png"
elif filename.endswith(".jpg") or filename.endswith(".jpeg"):
content_type = "image/jpeg"
elif filename.endswith(".gif"):
content_type = "image/gif"
headers = (
f"HTTP/1.1 {status_code} {status_message}\r\n"
f"Server: PyWebServer/amethyst-build-{AMETHYST_BUILD_NUMBER}\r\n"
f"Server: PyWebServer/1.0\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
f"Connection: close\r\n\r\n" # connection close bcuz im lazy
)
return headers.encode() + binary_data
def build_response(self, status_code, body):
"""
For textfiles we'll not have to guess MIME-types, though the other function
build_binary_response will be merged in here anyway.
"""
@staticmethod
def build_response(status_code, body):
messages = {
200: "OK",
204: "No Content",
302: "Found",
304: "Not Modified", # TODO KEKL
400: "Bad Request",
403: "Forbidden",
404: "Not Found",
405: "Method Not Allowed",
413: "Payload Too Large",
500: "Internal Server Error",
635: "Go Away"
}
status_message = messages.get(status_code)
if isinstance(body, str):
body = body.encode()
# TODO: dont encode yet, and i encode. awesome comments here.
# 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: PyWebServer/1.0\r\n"
f"Content-Length: {len(body)}\r\n"
f"Connection: close\r\n\r\n"
).encode()
if status_code == 302:
# 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}/"
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"Content-Length: {len(body)}\r\n"
f"Connection: close\r\n\r\n"
).encode()
return headers + body
def shutdown(self, signum, frame):
print("\nRecieved signal to exit!\nShutting down server...")
print(f"\nRecieved signal {signum}")
print("\nShutting down server...")
self.running = False
self.http_socket.close()
self.https_socket.close()
@@ -637,25 +417,12 @@ class WebServer:
def main():
print(
"WARNING!!\n"
f"This is Amethyst alpha build {AMETHYST_BUILD_NUMBER}\n"
"Since this is an alpha version of Amethyst, most features aren't working!\n"
"These builds are also very verbose and will spit out a lot on the terminal. "
"As you can imagine, this is for debugging purposes.\n"
"THERE IS ABSOLUTELY NO SUPPORT FOR THESE VERSIONS!\n"
"DO NOT USE THEM IN PRODUCTION SETTINGS!\n"
f"Please report any bugs on {AMETHYST_REPO}\n"
)
input("Press <Enter> to continue. ")
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
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
print(https_enabled)
file_handler.check_first_run()
http_port = file_handler.read_config("port") or 8080
https_port = file_handler.read_config("port-https") or 8443
http_enabled = file_handler.read_config("http") or True
https_enabled = file_handler.read_config("https") or False
server = WebServer(http_port=http_port, https_port=https_port)
server.start(http_enabled, https_enabled)