forked from pastagang/rigatoni
117 lines
3.5 KiB
Python
117 lines
3.5 KiB
Python
import socket
|
|
import ssl
|
|
|
|
SUPPORTED_SCHEMES = ["http", "https", "file"]
|
|
|
|
#=====#
|
|
# URL #
|
|
#=====#
|
|
class URL:
|
|
def __init__(self, url):
|
|
self.scheme, url = url.split("://", 1)
|
|
assert self.scheme in SUPPORTED_SCHEMES
|
|
match self.scheme:
|
|
case "http":
|
|
self.port = 80
|
|
case "https":
|
|
self.port = 443
|
|
if "/" not in url:
|
|
url = url + "/"
|
|
self.host, url = url.split("/", 1)
|
|
if ":" in self.host:
|
|
self.host, port = self.host.split(":", 1)
|
|
self.port = int(port)
|
|
self.path = "/" + url
|
|
|
|
def __str__(self):
|
|
port_part = ":" + str(self.port)
|
|
if self.scheme == "https" and self.port == 443:
|
|
port_part = ""
|
|
if self.scheme == "http" and self.port == 80:
|
|
port_part = ""
|
|
return self.scheme + "://" + self.host + port_part + self.path
|
|
|
|
def resolve(self, url):
|
|
if "://" in url: return URL(url)
|
|
if not url.startswith("/"):
|
|
dir, _ = self.path.rsplit("/", 1)
|
|
while url.startswith("../"):
|
|
_, url = url.split("/", 1)
|
|
if "/" in dir:
|
|
dir, _ = dir.rsplit("/", 1)
|
|
url = dir + "/" + url
|
|
if not url.startswith("/"):
|
|
dir, _ = self.path.rsplit("/", 1)
|
|
url = dir + "/" + url
|
|
if url.startswith("//"):
|
|
return URL(self.scheme + ":" + url)
|
|
else:
|
|
return URL(self.scheme + "://" + self.host + \
|
|
":" + str(self.port) + url)
|
|
|
|
#=========#
|
|
# REQUEST #
|
|
#=========#
|
|
def request(url):
|
|
match url.scheme:
|
|
case "http" | "https":
|
|
return request_http(url)
|
|
case "file":
|
|
return request_file(url)
|
|
|
|
def request_file(url):
|
|
try:
|
|
with open(url.path, 'r') as file:
|
|
content = file.read()
|
|
return "200", "", content
|
|
except:
|
|
print("Error")
|
|
|
|
def request_http(url, depth=0):
|
|
if depth > 5:
|
|
raise Exception("Too many redirects")
|
|
# Create a new socket
|
|
s = socket.socket(
|
|
family=socket.AF_INET,
|
|
type=socket.SOCK_STREAM,
|
|
proto=socket.IPPROTO_TCP
|
|
)
|
|
s.connect((url.host, url.port))
|
|
# If the scheme is https, we need to wrap the socket in ssl
|
|
if url.scheme == "https":
|
|
ctx = ssl.create_default_context()
|
|
s = ctx.wrap_socket(s, server_hostname=url.host)
|
|
# Set the request content
|
|
req = "GET {} HTTP/1.0\r\n".format(url.path) # format replaces {} with self.path
|
|
req += "Host: {}\r\n".format(url.host)
|
|
req += "\r\n"
|
|
s.send(req.encode("utf8"))
|
|
# Get the response
|
|
response = s.makefile("r", encoding="utf8", newline="\r\n")
|
|
statusline = response.readline()
|
|
version, status, explanation = statusline.split(" ", 2)
|
|
response_headers = {}
|
|
# Read the response headers
|
|
while True:
|
|
line = response.readline()
|
|
if line == "\r\n": break
|
|
header, value = line.split(":", 1)
|
|
response_headers[header.casefold()] = value.strip()
|
|
assert "transfer-encoding" not in response_headers
|
|
assert "content-encoding" not in response_headers
|
|
if status.startswith("3") and "location" in response_headers:
|
|
return request_http(url.resolve(response_headers["location"]), depth + 1)
|
|
content = response.read()
|
|
s.close()
|
|
return status, response_headers, content
|
|
|
|
#======#
|
|
# TEST #
|
|
#======#
|
|
if __name__ == "__main__":
|
|
import sys
|
|
status, headers, content = request(URL(sys.argv[1]))
|
|
print(content)
|
|
|
|
|
|
|