sniph-bank/main-doc.py

# Hi.
# I was tempted to implement my own custom HTTP server to show how an HTTP server works.
# But I'm pretty sure that's out of the scope of this demo.
# After all, HTTPS is just HTTP with TLS on top of it, and I'm not about to program Diffe-Hellman for this.
# Anyways, holy yap sesh.

# Import the HTTPStatus "enumeration" to have more readable code.
# An enumeration is just a thing with multiple diffent possibilities, and there are only so many HTTP
# status codes outlined by HTTP specifications.
# In reality, HTTP statuses are intentgers, with 1XX-5XX, with the first number dictating the category
# of error.
# For example, the HTTPStatus.NOT_FOUND would be a 404 error status code,
# And a HTTPStatus.OK would be the 200 success status code
from http import HTTPStatus

# This (flask) is the HTTP server library we are using.
# Response is the Python object representing what we return to the client
# Request is the Python object representing the request from the client
# send_file is a function that turns the contents of the specified file into a Response
# Flask is the server itself.
from flask import Flask, Response, request, send_file

# Cyrptographically secure, random bits.
# On Linux this is reading the bits from the special file /dev/urandom
# Less specifically, this uses a lot of the system's entropy (non-deterministic things based off practically
# random events, like the users recent mouse movements and secure positions) to generate random bits.
from os import urandom

# The threading library allows you to spawn multiple processes from within this one program.
# This is called concurrency, with each Thread being a concurrent process spun from a main thread.
# The Thread object represents one of those other Threads within the process.
# The Event object just allows for cross-thread communications.
# In this program, we will spin two threads up, one for the HTTP server and one for the HTTP server.
# We will use the Event object as a way to tell the servers to shut down.
from threading import Thread, Event

# The make_server function is just a sort of wrapper for the HTTP server to allow things like shutting down
# the server through a function call. Normally the call to run a Flask server is blocking I believe,
# meaning it stops program execution until it returns, which doesn't happen under normal circumstances.
from werkzeug.serving import make_server

# This is for type hinting. This is mostly for me, so that I know the type that each variable should be.
# Optional means that it can either be the None object, or whatever is within the brackets following the
# "Optional" type hint.
# For example, Optional[str] means that the value will either be None or a string.
from typing import Optional

# More type hinting...
from types import FrameType

# This is what allows us to handle signals like Ctrl+C (SIGINT), which we will use to trigger the Event
# for the Threads to know to shut the server down
import signal

# We only use this as a way to tell the system the program excuted sucessfully, and as intented.
import sys

# Used as a way to encode the password, or way of "hashing" it.
from base64 import b64encode as base64_encode

# The IP address the server will listen on.
SERVER_IP: str = "127.0.0.1"
# The exepected HTTP host header.
# This can use used to only response if the URL in the web browser contains "sniphbank.com"
# rather than accessing the IP (127.0.0.1) directly. Right now, this is not enabled.
HTTP_HOST: str = "sniphbank.com"
# The port for the HTTP socket.
# 80 is the default port, and what the browser automatically connects to when given "http://"
HTTP_PORT: int = 80
# The port for the HTTPS socket.
# 443 is the default port, and what the browser automatically connects to when given "https://"
HTTPS_PORT: int = 443
# This will be used to tell the Flask server whether or not it should be running in Debug mode.
DEBUG: bool = False

# Required to be in global scope
# The Flask server itself.
app = Flask(__name__)
# A secret key required for secure operations.
app.secret_key = urandom(24)


# The logic for what happens when someone visits the root of the HTTP server, so
# "http(s)://sniphbank.com", as the browser automatically requests "/"
@app.route("/")
def index():
    # We simply return the contents of the index.html file stored in the "website" directory in this
    # projects structure.
    return send_file("./website/index.html")


# The logic for when a user clicks on the "submit" button.
# The HTML login form's submit button sends an HTTP "POST" to "/login", and the response is then
# awaited for as if you clicked on a link to a new page.
@app.route("/login", methods=["POST"])
def login():
    # There isn't really a reason to validate the user for the demo,
    # just decide on a username that might pertain to whomever's
    # personal information

    # The "user" input field in the HTML form.
    user = request.form["username"]
    # Likewise, the "password" input field in the HTML form.
    # We turn this into bytes so we can encode it with base64
    password = request.form["password"].encode()

    # In a real web application, the values are pulled from a database and not hardcoded.
    # If the username is not "ronniej"
    if user != "ronniej":
        # Then we return the respponse "User not found" with the HTTP 401 status code (UNAUTHORIZED)
        return Response("User not found", HTTPStatus.UNAUTHORIZED)
    # Also in a real application, the password will be hashed rather than merely base64 encoded.
    # The difference with it being hashed is that it's easly reversable, someone with access to this
    # code (or in a real case, a database) would be able to easily reverse this to get the genuine
    # password.
    # The point of this here is to showcase that password is in plaintext until it gets to the server.
    if base64_encode(password).decode() != "Z29GQUxDT05TMTIzCg==":
        # Likewise if the password doesn't match, we return a similar response.
        return Response("Incorrect password", HTTPStatus.UNAUTHORIZED)

    # In order to get this far, the username and password are correct, so we can send the webpage.
    return send_file("./website/account.html")


# CSS, or Cascading Style Sheets, is the very thing that makes the web useable.
# Otherwise, every element in HTML (Hypertext Markup Language, NOT a programming language)
# would be completely black text on white background, and blocks placed top to bottom
# If you wanna try it yourself, feel free to run this code and rename/move/change the file path
# below. The server will return a "404: Not Found" error and the browser won't be able to render
# the webpage as expected.
@app.route("/css/style.css", methods=["GET"])
def stylesheet():
    return send_file("./website/css/style.css")


# Very similar, the favicon.ico is automatically requested from the server by the Web Browser
# This is the lil icon that is displayed on the tab next to the HTML page's title.
@app.route("/favicon.ico", methods=["GET"])
def favicon():
    return send_file("./website/favicon.ico")


# This is a lil fancy, but I thought it was a fair obstraction to handle the threading in a...more...
# "readable" way...?
# Basically, we are creating a class that is based off the "Thread" class, so it shares the same properties
# and methods.
# Here, we are going to add some things to it, for example, the Flask server, the IP and port we want
# to listen on, and the SSL Context if we are using HTTPS
class ServerThread(Thread):
    def __init__(
        self,
        app: Flask,
        host: str,
        port: int,
        ssl_context: Optional[tuple[str, str]] = None,
        name: str = "ServerThread",
    ) -> None:
        super().__init__(name=name, daemon=True)
        # threaded=True is telling the werkzeug server that it can use threads to handle multiple
        # requests at the same time. Not really important for the demo I don't think, and I'll
        # likely get rid of it.
        self.server = make_server(
            host,
            port,
            app,
            threaded=True,
            # This is either None or the public SSL cert and private key (or is it the other way around...?)
            ssl_context=ssl_context,  # tuple (cert, key) or SSLContext
        )

    def run(self):
        # serve_forever blocks until shutdown() is called
        self.server.serve_forever()

    def shutdown(self):
        self.server.shutdown()


def _install_signal_handlers(stop_event: Event):
    def handler(signum: signal.Signals, frame: Optional[FrameType]):
        stop_event.set()

    # Handle Ctrl+C and SIGTERM
    signal.signal(signal.SIGINT, handler)
    try:
        signal.signal(signal.SIGTERM, handler)
    except AttributeError:
        # SIGTERM may not be available on some platforms (e.g., Windows Python in some contexts)
        # I'm always on Linux, so if you wanna run this, then you should either use WSL or Podman.
        # (Podman is a generally more performant and secure alternative to Docker, if you are familiar)
        pass


# This block literally just says "If this is the program being run, do the following"
if __name__ == "__main__":
    # Create the stop event
    stop_event = Event()
    # Install the signal handler for SIGINT (Ctrl+C signal)
    _install_signal_handlers(stop_event)

    # Our HTTP server thread.
    http_server = ServerThread(
        app, SERVER_IP, HTTP_PORT, ssl_context=None, name="HTTP-Thread"
    )
    # Similarly, the HTTPS server thread.
    https_server = ServerThread(
        app,
        SERVER_IP,
        HTTPS_PORT,
        ssl_context=("./certs/cert.pem", "./certs/key.pem"),
        name="HTTPS-Thread",
    )

    try:
        # Start the HTTP and HTTPS server threads
        http_server.start()
        https_server.start()
        # Make sure there's some output to show that the servers running.
        print(f"HTTP  server running at http://{SERVER_IP}:{HTTP_PORT}")
        print(f"HTTPS server running at https://{SERVER_IP}:{HTTPS_PORT}")
        print("Press Ctrl+C to stop.")
        # Wait until a signal requests shutdown
        stop_event.wait()
    except KeyboardInterrupt:
        # Typically how you handle Ctrl+C, but since we have a singal hanlder it's pointless.
        # Doesn't hurt anyone though.
        pass
    finally:
        # Stop both servers.
        for srv in (http_server, https_server):
            try:
                srv.shutdown()
            except Exception:
                pass

        # Join both Threads (meaning take their execution and join it with the main thread's).
        for srv in (http_server, https_server):
            srv.join(timeout=5)

        # Output that the servers shut down gracefully (enough)
        print("Servers stopped!")
        # This just tells the shell that "hey, we exited as expected."
        sys.exit(0)