1from __future__ import annotations
2
3#
4#
5# This file is part of gunicorn released under the MIT license.
6# See the LICENSE for more information.
7#
8# Vendored and modified for Plain.
9import email.utils
10import fcntl
11import html
12import os
13import random
14import re
15import socket
16import time
17import urllib.parse
18from typing import Any
19
20# Server and Date aren't technically hop-by-hop
21# headers, but they are in the purview of the
22# origin server, so we drop them and add our own.
23#
24# In the future, concatenation server header values
25# might be better, but nothing else does it and
26# dropping them is easier.
27hop_headers = set(
28 """
29 connection keep-alive proxy-authenticate proxy-authorization
30 te trailers transfer-encoding upgrade
31 server date
32 """.split()
33)
34
35
36def is_ipv6(addr: str) -> bool:
37 try:
38 socket.inet_pton(socket.AF_INET6, addr)
39 except OSError: # not a valid address
40 return False
41 except ValueError: # ipv6 not supported on this platform
42 return False
43 return True
44
45
46def parse_address(netloc: str, default_port: str = "8000") -> str | tuple[str, int]:
47 if re.match(r"unix:(//)?", netloc):
48 return re.split(r"unix:(//)?", netloc)[-1]
49
50 if netloc.startswith("tcp://"):
51 netloc = netloc.split("tcp://")[1]
52 host, port = netloc, default_port
53
54 if "[" in netloc and "]" in netloc:
55 host = netloc.split("]")[0][1:]
56 port = (netloc.split("]:") + [default_port])[1]
57 elif ":" in netloc:
58 host, port = (netloc.split(":") + [default_port])[:2]
59 elif netloc == "":
60 host, port = "0.0.0.0", default_port
61
62 try:
63 port = int(port)
64 except ValueError:
65 raise RuntimeError(f"{port!r} is not a valid port number.")
66
67 return host.lower(), port
68
69
70def close_on_exec(fd: int) -> None:
71 flags = fcntl.fcntl(fd, fcntl.F_GETFD)
72 flags |= fcntl.FD_CLOEXEC
73 fcntl.fcntl(fd, fcntl.F_SETFD, flags)
74
75
76def _error_response_bytes(status_int: int, reason: str, mesg: str) -> bytes:
77 body = (
78 "<html>\n"
79 f" <head><title>{reason}</title></head>\n"
80 " <body>\n"
81 f" <h1><p>{reason}</p></h1>\n"
82 f" {html.escape(mesg)}\n"
83 " </body>\n"
84 "</html>\n"
85 )
86
87 response = (
88 f"HTTP/1.1 {status_int} {reason}\r\n"
89 f"Connection: close\r\n"
90 f"Content-Type: text/html\r\n"
91 f"Content-Length: {len(body)}\r\n"
92 f"\r\n"
93 f"{body}"
94 )
95 return response.encode("latin1")
96
97
98def http_date(timestamp: float | None = None) -> str:
99 """Return the current date and time formatted for a message header."""
100 if timestamp is None:
101 timestamp = time.time()
102 s = email.utils.formatdate(timestamp, localtime=False, usegmt=True)
103 return s
104
105
106def is_hoppish(header: str) -> bool:
107 return header.lower().strip() in hop_headers
108
109
110def seed() -> None:
111 try:
112 random.seed(os.urandom(64))
113 except NotImplementedError:
114 random.seed(f"{time.time()}.{os.getpid()}")
115
116
117def to_bytestring(value: str | bytes, encoding: str = "utf8") -> bytes:
118 """Converts a string argument to a byte string"""
119 if isinstance(value, bytes):
120 return value
121 if not isinstance(value, str):
122 raise TypeError(f"{value!r} is not a string")
123
124 return value.encode(encoding)
125
126
127def make_fail_handler(msg: str | bytes) -> Any:
128 """Create a handler that returns a 500 error for all requests."""
129 msg = to_bytestring(msg)
130
131 class FailHandler:
132 async def handle(self, request: Any, executor: Any) -> Any:
133 from plain.http import Response
134
135 return Response(msg, status_code=500, content_type="text/plain")
136
137 return FailHandler()
138
139
140def split_request_uri(uri: str) -> urllib.parse.SplitResult:
141 if uri.startswith("//"):
142 # When the path starts with //, urlsplit considers it as a
143 # relative uri while the RFC says we should consider it as abs_path
144 # http://www.w3.org/Protocols/rfc2616/rfc2616-sec5.html#sec5.1.2
145 # We use temporary dot prefix to workaround this behaviour
146 parts = urllib.parse.urlsplit("." + uri)
147 return parts._replace(path=parts.path[1:])
148
149 return urllib.parse.urlsplit(uri)
150
151
152def bytes_to_str(b: str | bytes) -> str:
153 if isinstance(b, str):
154 return b
155 return str(b, "latin1")