1124 lines
48 KiB
Python
1124 lines
48 KiB
Python
"""Networking and IT tools.
|
||
|
||
Two kinds live here. `subnet_calc` and `network_reference` are pure
|
||
computation against local data - no network, no rate limit, and they
|
||
cannot be wrong. The other four reach out and touch a host the user named,
|
||
so they carry a per-user hourly budget (see limits.py). squid is what makes
|
||
them safe; the budget is what makes them polite.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import csv
|
||
import datetime
|
||
import ipaddress
|
||
import logging
|
||
import re
|
||
import ssl
|
||
from functools import lru_cache
|
||
from pathlib import Path
|
||
from typing import Literal
|
||
|
||
from fastapi import APIRouter, Depends
|
||
from pydantic import Field
|
||
from starlette.concurrency import run_in_threadpool
|
||
|
||
from .. import cache
|
||
from ..httpc import FetchError, get_client, get_json, tls_socket
|
||
from ..limits import Identity, identity, spend_probe
|
||
from ..security import UnsafeURL, require_auth, validate_hostname, validate_url
|
||
from ..toolbase import ToolModel, tool
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
router = APIRouter(prefix="/tools", tags=["network"], dependencies=[Depends(require_auth)])
|
||
|
||
DATA_DIR = Path(__file__).resolve().parent.parent / "data"
|
||
|
||
|
||
# --------------------------------------------------------------- subnet_calc
|
||
|
||
class SubnetRequest(ToolModel):
|
||
network: str = Field(
|
||
description="CIDR, e.g. '192.168.1.0/24' or '2001:db8::/32'. A bare address is one host; "
|
||
"'address mask' form ('192.168.1.0 255.255.255.0') also works.",
|
||
)
|
||
split_into: int | None = Field(
|
||
default=None,
|
||
description="Divide the network into this many equal subnets (rounded up to a power of "
|
||
"two) and list them.",
|
||
)
|
||
split_prefix: int | None = Field(
|
||
default=None,
|
||
description="Carve the network into subnets of this prefix length, e.g. 56 to divide a /48 "
|
||
"into /56s. Use with `nth` to pick one without listing them all.",
|
||
)
|
||
nth: int | None = Field(
|
||
default=None,
|
||
description="Which subnet of the split to return, counting from 1 - 'the 4th /56 of this "
|
||
"/48'. Never work these out by hand: IPv6 prefix boundaries fall inside a "
|
||
"hextet.",
|
||
)
|
||
|
||
|
||
_SPECIAL_BLOCKS = (
|
||
("2001:db8::/32", "documentation only (RFC 3849) - not routable, and not a ULA"),
|
||
("fc00::/7", "unique local address, ULA (RFC 4193)"),
|
||
("192.0.2.0/24", "documentation only (RFC 5737, TEST-NET-1)"),
|
||
("198.51.100.0/24", "documentation only (RFC 5737, TEST-NET-2)"),
|
||
("203.0.113.0/24", "documentation only (RFC 5737, TEST-NET-3)"),
|
||
("100.64.0.0/10", "carrier-grade NAT shared address space (RFC 6598)"),
|
||
)
|
||
|
||
# The same fact as a sentence. Three graded rounds of one prompt showed the
|
||
# model reading "Scope: documentation only (RFC 5737)" as a field to skip
|
||
# and the advisory paragraph beside it as something to reproduce clause for
|
||
# clause; a fact the answer should carry has to be shaped like the prose it
|
||
# does carry.
|
||
_DOCUMENTATION_SENTENCE = (
|
||
"This prefix is reserved for documentation and examples ({rfc}): it is never allocated, "
|
||
"never announced on the public Internet, and hosts do not route to it. If it stands in "
|
||
"for a real prefix in a question, the arithmetic is the same for the real one - say so, "
|
||
"and answer for the real prefix."
|
||
)
|
||
|
||
_RESERVED_KINDS = (
|
||
("10.0.0.0/8", "private address space (RFC 1918)"),
|
||
("172.16.0.0/12", "private address space (RFC 1918)"),
|
||
("192.168.0.0/16", "private address space (RFC 1918)"),
|
||
("127.0.0.0/8", "the loopback range (RFC 1122)"),
|
||
("169.254.0.0/16", "link-local address space (RFC 3927)"),
|
||
("0.0.0.0/8", "\"this network\" (RFC 1122), not a host address"),
|
||
("224.0.0.0/4", "multicast address space (RFC 5771)"),
|
||
("240.0.0.0/4", "reserved for future use (RFC 1112)"),
|
||
("::1/128", "the IPv6 loopback address"),
|
||
("fe80::/10", "IPv6 link-local address space (RFC 4291)"),
|
||
("ff00::/8", "IPv6 multicast address space (RFC 4291)"),
|
||
)
|
||
|
||
|
||
def _describe_reserved(ip) -> tuple[str, str]:
|
||
"""(block, description) for an address that has no public registration."""
|
||
for cidr, label in _SPECIAL_BLOCKS + _RESERVED_KINDS:
|
||
block = ipaddress.ip_network(cidr)
|
||
if block.version == ip.version and ip in block:
|
||
return cidr, label
|
||
return "a reserved range", "reserved or special-purpose address space (IANA special-purpose registry)"
|
||
|
||
|
||
def _special_use(net) -> str | None:
|
||
"""Name the specific reserved block rather than lumping it under "private".
|
||
|
||
`is_private` is true for the documentation prefixes, so 2001:db8::/32 was
|
||
being reported as "private (RFC 1918 / ULA)" - wrong on both counts, and
|
||
misleading in the direction that matters: it is not assignable at all.
|
||
"""
|
||
for cidr, label in _SPECIAL_BLOCKS:
|
||
block = ipaddress.ip_network(cidr)
|
||
if block.version == net.version and net.subnet_of(block):
|
||
return label
|
||
return None
|
||
|
||
|
||
def _slaac_note(version: int, prefixlen: int, *, splitting: bool) -> str | None:
|
||
"""Warn when a prefix longer than /64 is described or produced.
|
||
|
||
Asked to carve a /64 into four VLANs, this tool returned eight /67s with
|
||
perfect arithmetic and no comment, while the same output carried a
|
||
"Scope: documentation only (RFC 3849)" line. The answer that followed
|
||
tracked the tool one for one - it repeated the documentation caveat the
|
||
tool flagged, said nothing about the /64 boundary the tool did not, and
|
||
recommended SLAAC on /67s. The advisory surface of this tool is what
|
||
decides which facts reach the answer, so the boundary belongs here rather
|
||
than in a prompt.
|
||
|
||
IPv4 has no such boundary, and /64 or shorter is exactly what SLAAC wants,
|
||
so neither is flagged. A single /128 being *described* is a host address or
|
||
a host route rather than a subnet, so it is not flagged either; a split
|
||
that produces /128s is.
|
||
"""
|
||
if version != 6 or prefixlen <= 64:
|
||
return None
|
||
if splitting is False and prefixlen == 128:
|
||
return None
|
||
subject = "The arithmetic below is correct" if splitting else "The details above are correct"
|
||
return (
|
||
f"Note: a /{prefixlen} is longer than /64. SLAAC needs a 64-bit interface identifier "
|
||
"(RFC 4291 section 2.5.1; RFC 7421 explains why the boundary is fixed), so hosts on a "
|
||
f"/{prefixlen} cannot autoconfigure an address - they need DHCPv6 or static addressing, "
|
||
"and some stacks (notably Android) ship no DHCPv6 client at all. "
|
||
f"{subject}; whether to subnet past /64 is a design decision. Where the constraint is "
|
||
"\"I only have one /64\", the usual answer is to ask the provider for a /56 or /48 by "
|
||
"DHCPv6-PD rather than to split the /64."
|
||
)
|
||
|
||
|
||
def _reverse_zone(net) -> str:
|
||
"""Reverse DNS zone for a *prefix*, not for its first address.
|
||
|
||
`reverse_pointer` on a network address gives the PTR name of that single
|
||
address. For 2001:db8:4f2:30b::/64 that is all 32 nibbles, sixteen of which
|
||
are the host part and belong to records *inside* the zone rather than to the
|
||
zone name. Delegation happens on the prefix, so the name has to be truncated
|
||
to it. IPv4 had the same bug and hid it better: a /24 returned
|
||
0.10.168.192.in-addr.arpa, which is the PTR of 192.168.1.0 rather than the
|
||
zone 10.168.192.in-addr.arpa - one label too many, and easy to read past.
|
||
|
||
Only prefixes on a nibble (v6) or octet (v4) boundary name a single zone;
|
||
anything else is delegated with RFC 2317 style CNAMEs, so say so rather than
|
||
inventing a zone that cannot exist.
|
||
"""
|
||
full = net.network_address.reverse_pointer
|
||
step = 4 if net.version == 6 else 8
|
||
suffix = "nibble" if net.version == 6 else "octet"
|
||
article = "a" if net.version == 6 else "an"
|
||
if net.prefixlen % step:
|
||
return (f"{full} [that is the PTR name of the network address itself. A "
|
||
f"/{net.prefixlen} is not on {article} {suffix} boundary, so it has no single "
|
||
"reverse zone - delegation needs RFC 2317 style CNAMEs.]")
|
||
labels = full.split(".")
|
||
keep = net.prefixlen // step
|
||
return ".".join(labels[len(labels) - 2 - keep:])
|
||
|
||
|
||
@router.post("/subnet_calc", operation_id="subnet_calc", response_model=str,
|
||
summary="Calculate subnet details")
|
||
@tool
|
||
async def subnet_calc(body: SubnetRequest) -> str:
|
||
"""Subnet details for IPv4 or IPv6: network, broadcast, usable range, mask, host count, splits.
|
||
|
||
Use it for every subnetting question rather than doing the arithmetic yourself -
|
||
off-by-one errors in usable ranges and broadcast addresses land in real firewall rules
|
||
and DHCP scopes. It also splits a network into equal subnets, which is what "divide
|
||
this /24 into N networks" wants.
|
||
"""
|
||
raw = body.network.strip()
|
||
# "192.168.1.0 255.255.255.0" is how a lot of gear prints it.
|
||
if " " in raw and "/" not in raw:
|
||
addr, _, mask = raw.partition(" ")
|
||
raw = f"{addr.strip()}/{mask.strip()}"
|
||
|
||
try:
|
||
net = ipaddress.ip_network(raw, strict=False)
|
||
except ValueError as exc:
|
||
return f"CANNOT DO THAT: '{body.network}' is not a valid network or address ({exc})."
|
||
|
||
given = None
|
||
if "/" not in body.network or net.prefixlen == net.max_prefixlen:
|
||
try:
|
||
given = ipaddress.ip_interface(raw)
|
||
except ValueError:
|
||
given = None
|
||
|
||
lines = [f"# {net.with_prefixlen}", ""]
|
||
|
||
if net.version == 4:
|
||
total = net.num_addresses
|
||
usable = max(total - 2, 0) if net.prefixlen < 31 else total
|
||
lines += [
|
||
f"Network address: {net.network_address}",
|
||
f"Broadcast: {net.broadcast_address}",
|
||
f"Netmask: {net.netmask} (/{net.prefixlen})",
|
||
f"Wildcard mask: {net.hostmask}",
|
||
f"Total addresses: {total:,}",
|
||
f"Usable hosts: {usable:,}",
|
||
]
|
||
if usable and net.prefixlen < 31:
|
||
# Deliberately arithmetic rather than list(net.hosts()): that
|
||
# materialises every address, and a /8 is 16.7 million of them.
|
||
# It OOM-killed the container the first time this was exercised.
|
||
first = ipaddress.ip_address(int(net.network_address) + 1)
|
||
last = ipaddress.ip_address(int(net.broadcast_address) - 1)
|
||
lines.append(f"Usable range: {first} - {last}")
|
||
elif net.prefixlen == 31:
|
||
lines.append(f"Usable range: {net.network_address} - {net.broadcast_address} (RFC 3021 point-to-point)")
|
||
else:
|
||
lines += [
|
||
f"Network address: {net.network_address}",
|
||
f"Prefix length: /{net.prefixlen}",
|
||
f"Total addresses: {net.num_addresses:,}",
|
||
f"Address range: {net.network_address} - {net[-1]}",
|
||
]
|
||
|
||
scope = []
|
||
if special := _special_use(net):
|
||
scope.append(special)
|
||
elif net.is_private:
|
||
scope.append("private (RFC 1918 / ULA)")
|
||
if net.is_global:
|
||
scope.append("globally routable")
|
||
if net.is_loopback:
|
||
scope.append("loopback")
|
||
if net.is_link_local:
|
||
scope.append("link-local")
|
||
if net.is_multicast:
|
||
scope.append("multicast")
|
||
lines.append(f"Scope: {', '.join(scope) or 'reserved/special-use'}")
|
||
if special and "documentation" in special:
|
||
rfc = re.search(r"RFC \d+", special)
|
||
lines += ["", _DOCUMENTATION_SENTENCE.format(rfc=rfc.group(0) if rfc else "documentation range")]
|
||
if note := _slaac_note(net.version, net.prefixlen, splitting=False):
|
||
lines += ["", note]
|
||
|
||
if given is not None and given.ip != net.network_address:
|
||
lines += ["", f"The address {given.ip} sits inside this network."]
|
||
|
||
try:
|
||
lines.append(f"Reverse DNS zone: {_reverse_zone(net)}")
|
||
except Exception:
|
||
pass
|
||
|
||
if body.split_prefix is not None:
|
||
want = body.split_prefix
|
||
if want <= net.prefixlen or want > net.max_prefixlen:
|
||
lines += ["", f"Cannot carve a /{net.prefixlen} into /{want} subnets - the target "
|
||
f"prefix must be longer than /{net.prefixlen} and no longer than "
|
||
f"/{net.max_prefixlen}."]
|
||
else:
|
||
bits = want - net.prefixlen
|
||
count = 2 ** bits
|
||
lines += ["", f"## Carved into /{want} subnets"]
|
||
if note := _slaac_note(net.version, want, splitting=True):
|
||
lines += [note, ""]
|
||
lines.append(f"A /{net.prefixlen} contains 2^({want}-{net.prefixlen}) = 2^{bits} = "
|
||
f"**{count:,}** subnets of /{want}.")
|
||
if body.nth is not None:
|
||
n = body.nth
|
||
if n < 1 or n > count:
|
||
lines.append(f"There is no {n}th /{want} here - there are only {count:,}.")
|
||
else:
|
||
sub = ipaddress.ip_network(
|
||
(int(net.network_address) + (n - 1) * (2 ** (net.max_prefixlen - want)), want))
|
||
lines += [
|
||
"",
|
||
f"**#{n} (counting from 1) = {sub.with_prefixlen}**",
|
||
f"First address: {sub.network_address}",
|
||
f"Last address: {sub[-1]}",
|
||
f"Reverse zone: {_reverse_zone(sub)}",
|
||
"",
|
||
f"Index {n} is offset {n - 1} from the base, and each /{want} step is "
|
||
f"2^({net.max_prefixlen}-{want}) = 2^{net.max_prefixlen - want} addresses. "
|
||
f"If you meant zero-based numbering, #{n} zero-based is "
|
||
f"{ipaddress.ip_network((int(net.network_address) + n * (2 ** (net.max_prefixlen - want)), want)).with_prefixlen if n < count else 'past the end'}"
|
||
" - say which convention you used.",
|
||
]
|
||
else:
|
||
shown = [x for x, _ in zip(net.subnets(new_prefix=want), range(16))]
|
||
lines.append("")
|
||
for i, sub in enumerate(shown, 1):
|
||
lines.append(f"- #{i}: {sub.with_prefixlen}")
|
||
if count > 16:
|
||
lines.append(f"- ... and {count - 16:,} more. Pass nth=N to get one directly "
|
||
"rather than counting down this list.")
|
||
|
||
elif body.split_into and body.split_into > 1:
|
||
import math
|
||
|
||
bits = math.ceil(math.log2(body.split_into))
|
||
new_prefix = net.prefixlen + bits
|
||
if new_prefix > net.max_prefixlen:
|
||
lines += ["", f"Cannot split into {body.split_into} subnets - that would need a "
|
||
f"/{new_prefix}, which is longer than /{net.max_prefixlen}."]
|
||
else:
|
||
count = 2 ** bits
|
||
generator = net.subnets(prefixlen_diff=bits)
|
||
shown = [s for s, _ in zip(generator, range(32))]
|
||
lines += ["", f"## Split into {count:,} × /{new_prefix} subnets"]
|
||
if note := _slaac_note(net.version, new_prefix, splitting=True):
|
||
lines += [note, ""]
|
||
lines += [f"(you asked for {body.split_into}; subnetting works in powers of two)", ""]
|
||
for sub in shown:
|
||
if sub.version == 4 and sub.prefixlen < 31:
|
||
first = ipaddress.ip_address(int(sub.network_address) + 1)
|
||
last = ipaddress.ip_address(int(sub.broadcast_address) - 1)
|
||
lines.append(f"- {sub.with_prefixlen} hosts {first} - {last} "
|
||
f"broadcast {sub.broadcast_address}")
|
||
else:
|
||
lines.append(f"- {sub.with_prefixlen}")
|
||
if count > 32:
|
||
lines.append(f"- … and {count - 32:,} more")
|
||
|
||
return "\n".join(lines)
|
||
|
||
|
||
# ---------------------------------------------------------------- dns_lookup
|
||
|
||
DNS_TYPES = Literal["A", "AAAA", "MX", "TXT", "NS", "CNAME", "SOA", "CAA", "SRV", "PTR"]
|
||
|
||
|
||
class DnsRequest(ToolModel):
|
||
name: str = Field(description="Domain name, e.g. 'example.com'. Public names only.")
|
||
record_type: DNS_TYPES = Field(
|
||
default="A",
|
||
description="Default A. A/AAAA=addresses, MX=mail servers, TXT=SPF/DKIM/verification, "
|
||
"NS=nameservers, CNAME=aliases, SOA=zone authority, CAA=certificate authority "
|
||
"authorisation, SRV=service records, PTR=reverse lookup.",
|
||
)
|
||
|
||
|
||
@router.post("/dns_lookup", operation_id="dns_lookup", response_model=str,
|
||
summary="Look up DNS records")
|
||
@tool
|
||
async def dns_lookup(body: DnsRequest, who: Identity = Depends(identity)) -> str:
|
||
"""Query live DNS records for a public domain name.
|
||
|
||
Use it to see what a domain resolves to, diagnose mail delivery (MX, SPF/DMARC in
|
||
TXT), confirm delegation, or check that a change has propagated. It queries the public
|
||
internet from this server and cannot see the user's local or split-horizon DNS - say
|
||
so when relevant.
|
||
"""
|
||
await spend_probe(who, "dns_lookup")
|
||
name = validate_hostname(body.name)
|
||
rtype = body.record_type
|
||
|
||
if hit := await cache.cached("dns", name, rtype):
|
||
return hit
|
||
|
||
payload = None
|
||
for endpoint, params in (
|
||
("https://dns.google/resolve", {"name": name, "type": rtype}),
|
||
("https://cloudflare-dns.com/dns-query", {"name": name, "type": rtype}),
|
||
):
|
||
try:
|
||
payload = await get_json(endpoint, params=params,
|
||
headers={"Accept": "application/dns-json"})
|
||
if payload:
|
||
break
|
||
except Exception as exc:
|
||
log.info("DoH resolver %s failed: %s", endpoint, exc)
|
||
|
||
if payload is None:
|
||
raise FetchError("No DNS resolver responded. Do not guess what the records contain.")
|
||
|
||
status = payload.get("Status", -1)
|
||
if status == 3:
|
||
return (f"NXDOMAIN — the domain '{name}' does not exist in public DNS. "
|
||
"Do not speculate about what it might resolve to.")
|
||
if status != 0:
|
||
return f"DNS query for {name} failed with status code {status}."
|
||
|
||
answers = payload.get("Answer") or []
|
||
records = [a for a in answers if _dns_type_name(a.get("type")) == rtype]
|
||
|
||
lines = [f"# DNS {rtype} records for {name}", ""]
|
||
if not records:
|
||
other = {_dns_type_name(a.get("type")) for a in answers}
|
||
if other:
|
||
lines.append(f"No {rtype} records exist. The name does resolve, but only has: "
|
||
f"{', '.join(sorted(t for t in other if t))}.")
|
||
else:
|
||
lines.append(f"No {rtype} records exist for this name.")
|
||
else:
|
||
for record in records:
|
||
ttl = record.get("TTL")
|
||
value = (record.get("data") or "").strip()
|
||
lines.append(f"- {value} (TTL {ttl}s)")
|
||
|
||
if authority := payload.get("Authority"):
|
||
soa = next((a.get("data", "") for a in authority if _dns_type_name(a.get("type")) == "SOA"), "")
|
||
if soa:
|
||
lines += ["", f"Authority: {soa}"]
|
||
|
||
lines += ["", "[Resolved from the public internet, not from the user's local network.]"]
|
||
result = "\n".join(lines)
|
||
await cache.store("dns", result, name, rtype)
|
||
return result
|
||
|
||
|
||
_DNS_TYPE_MAP = {1: "A", 2: "NS", 5: "CNAME", 6: "SOA", 12: "PTR", 15: "MX",
|
||
16: "TXT", 28: "AAAA", 33: "SRV", 257: "CAA"}
|
||
|
||
|
||
def _dns_type_name(code) -> str:
|
||
return _DNS_TYPE_MAP.get(code, str(code))
|
||
|
||
|
||
# --------------------------------------------------------------- inspect_tls
|
||
|
||
class TlsRequest(ToolModel):
|
||
hostname: str = Field(description="Hostname, e.g. 'example.com', without https://.")
|
||
port: int = Field(default=443, ge=1, le=65535,
|
||
description="TLS port, default 443.")
|
||
|
||
|
||
@router.post("/inspect_tls", operation_id="inspect_tls", response_model=str,
|
||
summary="Inspect a TLS certificate")
|
||
@tool
|
||
async def inspect_tls(body: TlsRequest, who: Identity = Depends(identity)) -> str:
|
||
"""Examine the live TLS certificate a public host serves.
|
||
|
||
Reports expiry, issuer, covered hostnames, chain validity, and the negotiated TLS
|
||
version and cipher. Use it for "is my certificate about to expire", "why does my
|
||
browser complain", "did my renewal deploy". Expired and self-signed certificates are
|
||
reported, not refused.
|
||
"""
|
||
await spend_probe(who, "inspect_tls")
|
||
host = validate_hostname(body.hostname)
|
||
port = body.port
|
||
if port not in (443, 80):
|
||
# squid's CONNECT allowlist would refuse anything else anyway; saying
|
||
# so plainly beats a confusing tunnel error.
|
||
return (f"CANNOT DO THAT: only port 443 can be inspected from this server (you asked "
|
||
f"for {port}). The egress proxy does not permit other ports.")
|
||
|
||
if hit := await cache.cached("tls", host, str(port)):
|
||
return hit
|
||
|
||
result = await run_in_threadpool(_inspect_tls_sync, host, port)
|
||
await cache.store("tls", result, host, str(port))
|
||
return result
|
||
|
||
|
||
def _inspect_tls_sync(host: str, port: int) -> str:
|
||
from cryptography import x509
|
||
from cryptography.hazmat.primitives import hashes
|
||
|
||
trusted = True
|
||
trust_error = ""
|
||
try:
|
||
wrapped, raw_sock = tls_socket(host, port, verify=True)
|
||
except ssl.SSLCertVerificationError as exc:
|
||
trusted = False
|
||
trust_error = exc.verify_message or str(exc)
|
||
wrapped, raw_sock = tls_socket(host, port, verify=False)
|
||
|
||
try:
|
||
der = wrapped.getpeercert(binary_form=True)
|
||
version = wrapped.version()
|
||
cipher = wrapped.cipher()
|
||
finally:
|
||
try:
|
||
wrapped.close()
|
||
finally:
|
||
raw_sock.close()
|
||
|
||
if not der:
|
||
raise FetchError(f"{host} did not present a certificate.")
|
||
|
||
cert = x509.load_der_x509_certificate(der)
|
||
now = datetime.datetime.now(datetime.UTC)
|
||
not_after = cert.not_valid_after_utc
|
||
not_before = cert.not_valid_before_utc
|
||
days_left = (not_after - now).days
|
||
|
||
try:
|
||
san = cert.extensions.get_extension_for_class(x509.SubjectAlternativeName)
|
||
names = san.value.get_values_for_type(x509.DNSName)
|
||
except x509.ExtensionNotFound:
|
||
names = []
|
||
|
||
lines = [f"# TLS certificate for {host}:{port}", ""]
|
||
|
||
if days_left < 0:
|
||
lines.append(f"**EXPIRED {abs(days_left)} days ago.**")
|
||
elif days_left <= 14:
|
||
lines.append(f"**EXPIRES IN {days_left} DAYS — renewal is urgent.**")
|
||
elif days_left <= 30:
|
||
lines.append(f"**Expires in {days_left} days.**")
|
||
else:
|
||
lines.append(f"Valid for another {days_left} days.")
|
||
|
||
if not trusted:
|
||
lines.append(f"**Chain does NOT validate:** {trust_error}. A browser will warn on this.")
|
||
else:
|
||
lines.append("Certificate chain validates against the public trust store.")
|
||
|
||
lines += [
|
||
"",
|
||
f"Subject: {cert.subject.rfc4514_string()}",
|
||
f"Issuer: {cert.issuer.rfc4514_string()}",
|
||
f"Valid from: {not_before:%Y-%m-%d %H:%M UTC}",
|
||
f"Valid until: {not_after:%Y-%m-%d %H:%M UTC}",
|
||
f"Serial: {cert.serial_number:x}",
|
||
f"Signature: {cert.signature_algorithm_oid._name}",
|
||
f"SHA-256: {cert.fingerprint(hashes.SHA256()).hex(':')}",
|
||
"",
|
||
f"Negotiated: {version}, cipher {cipher[0] if cipher else 'unknown'}",
|
||
]
|
||
|
||
if names:
|
||
shown = ", ".join(names[:25])
|
||
more = f" (+{len(names) - 25} more)" if len(names) > 25 else ""
|
||
lines += ["", f"Covers {len(names)} name(s): {shown}{more}"]
|
||
|
||
return "\n".join(lines)
|
||
|
||
|
||
# -------------------------------------------------------------- http_headers
|
||
|
||
class HeadersRequest(ToolModel):
|
||
url: str = Field(description="Full URL, e.g. https://example.com/path.")
|
||
|
||
|
||
_SECURITY_HEADERS = {
|
||
"strict-transport-security": "HSTS",
|
||
"content-security-policy": "CSP",
|
||
"x-frame-options": "clickjacking protection",
|
||
"x-content-type-options": "MIME sniffing protection",
|
||
"referrer-policy": "referrer policy",
|
||
"permissions-policy": "permissions policy",
|
||
}
|
||
|
||
|
||
def _hsts_detail(value: str) -> str:
|
||
"""Break an HSTS header into the three things people ask about.
|
||
|
||
The question is almost always "what is the max-age, and are
|
||
includeSubDomains and preload set" - and `max-age=0`, which switches HSTS
|
||
off, used to render identically to a two-year preloaded policy because the
|
||
tool only reported that the header existed.
|
||
|
||
Directives are matched as whole tokens between semicolons, case-insensitively.
|
||
A substring search for "preload" or "includesubdomains" also matches those
|
||
words inside another directive's value (a report URI, for instance), which
|
||
would report a policy the server never set.
|
||
"""
|
||
tokens = [token.strip() for token in value.split(";") if token.strip()]
|
||
names = {token.partition("=")[0].strip().lower() for token in tokens}
|
||
|
||
max_age: int | None = None
|
||
for token in tokens:
|
||
name, sep, raw = token.partition("=")
|
||
if name.strip().lower() == "max-age" and sep:
|
||
try:
|
||
max_age = int(raw.strip().strip('"'))
|
||
except ValueError:
|
||
max_age = None
|
||
break
|
||
|
||
if max_age is None:
|
||
age = "max-age: absent or unparseable - without a max-age the header sets no policy"
|
||
elif max_age == 0:
|
||
age = "max-age=0 - HSTS is switched off here, which tells browsers to forget the policy"
|
||
elif max_age < 3600:
|
||
age = f"max-age={max_age} (~{max_age // 60} minutes)"
|
||
elif max_age < 86400:
|
||
age = f"max-age={max_age} (~{max_age / 3600:.0f} hours)"
|
||
else:
|
||
age = f"max-age={max_age} (~{max_age / 86400:.0f} days)"
|
||
|
||
subdomains = "includesubdomains" in names
|
||
detail = (f"{age}; includeSubDomains: {'yes' if subdomains else 'no'}; "
|
||
f"preload: {'yes' if 'preload' in names else 'no'}")
|
||
if "preload" in names and (max_age is None or max_age < 31536000 or not subdomains):
|
||
detail += (". The preload directive is set but the policy does not meet the preload "
|
||
"list's own requirements (max-age of at least 31536000 plus includeSubDomains), "
|
||
"so a submission to hstspreload.org would be rejected as it stands")
|
||
return detail
|
||
|
||
|
||
def _clip(value: str, limit: int) -> str:
|
||
return value if len(value) <= limit else f"{value[:limit]}… [{len(value)} chars total]"
|
||
|
||
|
||
def _security_section(response) -> list[str]:
|
||
"""Report each security header's value, not merely that it exists.
|
||
|
||
Asked for a site's HSTS max-age, this tool answered "Present: HSTS, CSP,
|
||
clickjacking protection" - a membership test, `header in response.headers`,
|
||
that never emitted the value; none of these six appear in the raw-header
|
||
dump either, so the value the model had already fetched was nowhere in the
|
||
output. It noticed, said so, and spent five more tool calls chasing it.
|
||
|
||
The tool follows redirects and only the final response's headers were ever
|
||
read. Apex and www are configured separately in the real world and do
|
||
differ: cloudflare.com sends max-age=15780000, www.cloudflare.com sends
|
||
max-age=31536000. The final response is the primary answer; an earlier hop
|
||
that disagrees is named alongside it rather than discarded.
|
||
"""
|
||
present: list[str] = []
|
||
missing: list[str] = []
|
||
notes: list[str] = []
|
||
|
||
for header, label in _SECURITY_HEADERS.items():
|
||
value = response.headers.get(header)
|
||
if value is None:
|
||
# "referrer-policy (referrer policy)" says nothing twice.
|
||
missing.append(header if label.lower() == header.replace("-", " ")
|
||
else f"{header} ({label})")
|
||
else:
|
||
present.append(f"{header}: {_clip(value, 300)}")
|
||
if header == "strict-transport-security":
|
||
present.append(f" {_hsts_detail(value)}")
|
||
|
||
# One note per distinct earlier value, naming the host that sent it.
|
||
earlier: dict[str, object] = {}
|
||
for hop in response.history:
|
||
hop_value = hop.headers.get(header)
|
||
if hop_value is not None and hop_value != value:
|
||
earlier.setdefault(hop_value, hop.url)
|
||
for other, hop_url in earlier.items():
|
||
hop_host = hop_url.host
|
||
if value is None:
|
||
notes.append(
|
||
f"Note: {hop_host} sends {header} ({_clip(other, 120)}) but the final "
|
||
f"response from {response.url.host} does not send it at all - they are "
|
||
"configured separately; report whichever the question is about, or both.")
|
||
elif hop_host != response.url.host:
|
||
notes.append(
|
||
f"Note: {header} differs between {hop_host} ({_clip(other, 120)}) and "
|
||
f"{response.url.host} ({_clip(value, 120)}) - the apex and www are "
|
||
"configured separately; report whichever the question is about, or both.")
|
||
else:
|
||
notes.append(
|
||
f"Note: {header} differs between the earlier hop {hop_url} "
|
||
f"({_clip(other, 120)}) and the final response {response.url} "
|
||
f"({_clip(value, 120)}) on the same host.")
|
||
|
||
if response.history and response.history[0].url.scheme != response.url.scheme:
|
||
notes.append(
|
||
f"Note: the scheme changed while following redirects - the request began at "
|
||
f"{response.history[0].url} and these headers are {response.url}'s. The "
|
||
"plain-http address itself answers with a redirect only, and an HSTS policy sent "
|
||
"over http is ignored by browsers (RFC 6797 section 8.1), so this is a statement "
|
||
"about the https address.")
|
||
|
||
lines = ["", f"## Security headers (final response: {response.url})"]
|
||
if present:
|
||
lines += present
|
||
else:
|
||
lines.append("None of the six security headers is present on the final response.")
|
||
lines.append(f"Missing: {', '.join(missing) if missing else 'none'}")
|
||
return lines + notes
|
||
|
||
|
||
@router.post("/http_headers", operation_id="http_headers", response_model=str,
|
||
summary="Inspect HTTP response headers")
|
||
@tool
|
||
async def http_headers(body: HeadersRequest, who: Identity = Depends(identity)) -> str:
|
||
"""A URL's HTTP status, redirect chain and response headers, without reading the page.
|
||
|
||
Use it to diagnose redirects and loops, check caching and CORS, see which server or
|
||
CDN answers, or read a site's security headers - each value is reported, with HSTS
|
||
broken into max-age, includeSubDomains and preload. For the page's content use
|
||
fetch_page.
|
||
"""
|
||
await spend_probe(who, "http_headers")
|
||
url = validate_url(body.url)
|
||
|
||
client = get_client()
|
||
response = await client.get(url, follow_redirects=True)
|
||
|
||
lines = [f"# {url}", ""]
|
||
|
||
if response.history:
|
||
lines.append("## Redirect chain")
|
||
for hop in response.history:
|
||
lines.append(f"{hop.status_code} {hop.url}\n -> {hop.headers.get('location', '')}")
|
||
lines += [f"{response.status_code} {response.url} (final)", ""]
|
||
else:
|
||
lines += [f"Status: {response.status_code} {response.reason_phrase}", ""]
|
||
|
||
interesting = [
|
||
"server", "content-type", "content-length", "content-encoding", "cache-control",
|
||
"expires", "etag", "last-modified", "location", "set-cookie", "x-powered-by",
|
||
"cf-ray", "via", "age", "vary", "access-control-allow-origin",
|
||
]
|
||
lines.append("## Response headers")
|
||
for key in interesting:
|
||
if value := response.headers.get(key):
|
||
lines.append(f"{key}: {value[:200]}")
|
||
|
||
lines += _security_section(response)
|
||
|
||
return "\n".join(lines)
|
||
|
||
|
||
# --------------------------------------------------------- lookup_ip_or_asn
|
||
|
||
class RdapRequest(ToolModel):
|
||
target: str = Field(
|
||
description="A public IP address (8.8.8.8), an AS number (AS15169 or 15169), or a domain "
|
||
"name.",
|
||
)
|
||
|
||
|
||
@router.post("/lookup_ip_or_asn", operation_id="lookup_ip_or_asn", response_model=str,
|
||
summary="Look up IP, ASN or domain registration")
|
||
@tool
|
||
async def lookup_ip_or_asn(body: RdapRequest, who: Identity = Depends(identity)) -> str:
|
||
"""Who a public IP address, AS number or domain is registered to, via RDAP.
|
||
|
||
The replacement for whois: the owner of an address seen in logs, an abuse contact,
|
||
which provider or cloud an address belongs to, a domain's registration and expiry
|
||
dates. Private addresses have no registration.
|
||
"""
|
||
await spend_probe(who, "lookup_ip_or_asn")
|
||
target = body.target.strip()
|
||
if not target:
|
||
return "CANNOT DO THAT: provide an IP address, AS number or domain name."
|
||
|
||
if hit := await cache.cached("rdap", target.lower()):
|
||
return hit
|
||
|
||
asn_match = re.fullmatch(r"(?:AS)?(\d{1,10})", target, re.I)
|
||
try:
|
||
ip = ipaddress.ip_address(target)
|
||
except ValueError:
|
||
ip = None
|
||
|
||
if ip is not None:
|
||
if not ip.is_global:
|
||
# "Private, loopback or reserved" was true and useless: asked
|
||
# about two TEST-NET addresses, the model knew from memory that
|
||
# they were documentation ranges, got this line back, and
|
||
# talked itself out of saying so. Name the block.
|
||
block, label = _describe_reserved(ip)
|
||
sentence = ""
|
||
if "documentation" in label:
|
||
rfc = re.search(r"RFC \d+", label)
|
||
sentence = " " + _DOCUMENTATION_SENTENCE.format(rfc=rfc.group(0) if rfc else "documentation range")
|
||
return (f"CANNOT DO THAT: {ip} is in {block}, {label}. It has no public registration "
|
||
f"and no origin AS to look up.{sentence}"
|
||
+ ("" if sentence else " It belongs to whoever runs that local network."))
|
||
url, kind = f"https://rdap.org/ip/{ip}", "IP address"
|
||
elif asn_match:
|
||
url, kind = f"https://rdap.org/autnum/{asn_match.group(1)}", "AS number"
|
||
else:
|
||
host = validate_hostname(target)
|
||
# RDAP registers the registrable domain, not arbitrary subdomains.
|
||
parts = host.split(".")
|
||
domain = ".".join(parts[-2:]) if len(parts) > 2 else host
|
||
url, kind = f"https://rdap.org/domain/{domain}", "domain"
|
||
|
||
routing = ""
|
||
if kind == "IP address":
|
||
routing = await _routing_view(str(ip))
|
||
elif kind == "AS number":
|
||
routing = await _as_view(asn_match.group(1))
|
||
|
||
try:
|
||
data = await get_json(url)
|
||
except Exception:
|
||
if routing:
|
||
result = (f"# {target} ({kind})\n\nNo RDAP registration record was returned for it - "
|
||
"some registries do not publish RDAP for every resource, so registration "
|
||
f"details are unknown; do not invent them.\n\n{routing}")
|
||
await cache.store("rdap", result, target.lower())
|
||
return result
|
||
return (f"NO RESULTS. No RDAP registration found for '{target}'. Some registries do not "
|
||
"publish RDAP for every resource. Do not invent ownership details.")
|
||
|
||
result = _format_rdap(target, kind, data)
|
||
if routing:
|
||
result += "\n\n" + routing
|
||
await cache.store("rdap", result, target.lower())
|
||
return result
|
||
|
||
|
||
# Registration is not routing. RDAP says who an address block is allocated
|
||
# to; which network actually announces it, and under which AS number, is a
|
||
# BGP question that RDAP cannot answer - and it is the question "which AS
|
||
# does this site sit in" asks. Two graded rounds of that prompt cost four
|
||
# extra fetches each for the origin AS, from whichever third-party page the
|
||
# model could find; RIPEstat publishes it from the collectors' routing view.
|
||
_RIPESTAT = "https://stat.ripe.net/data"
|
||
|
||
|
||
async def _routing_view(ip: str) -> str:
|
||
url = f"{_RIPESTAT}/prefix-overview/data.json?resource={ip}"
|
||
try:
|
||
payload = await get_json(url, timeout=12.0)
|
||
data = payload.get("data") or {}
|
||
except Exception:
|
||
return ("## Routing (BGP)\nOrigin AS: not available - RIPEstat did not answer. Do not "
|
||
"supply one from memory.")
|
||
lines = ["## Routing (BGP)"]
|
||
if not data.get("announced"):
|
||
lines.append(f"{ip} is not announced in BGP - no origin AS; the address is not reachable "
|
||
"on the public Internet right now (or is inside a block that is only "
|
||
"registered, not routed).")
|
||
else:
|
||
asns = data.get("asns") or []
|
||
prefix = data.get("resource") or "?"
|
||
origins = ", ".join(f"AS{a.get('asn')} ({a.get('holder') or 'holder unknown'})" for a in asns)
|
||
lines.append(f"Announced prefix: {prefix}")
|
||
lines.append(f"Origin AS: {origins or 'unknown'}")
|
||
if len(asns) > 1:
|
||
lines.append("More than one origin: the prefix is announced from several networks "
|
||
"(anycast or a multi-origin arrangement) - report all of them.")
|
||
lines.append(f"From RIPEstat's routing view (RIS collectors), {url}")
|
||
return "\n".join(lines)
|
||
|
||
|
||
async def _as_view(asn: str) -> str:
|
||
url = f"{_RIPESTAT}/as-overview/data.json?resource=AS{asn}"
|
||
try:
|
||
payload = await get_json(url, timeout=12.0)
|
||
data = payload.get("data") or {}
|
||
except Exception:
|
||
return ""
|
||
holder = data.get("holder") or "unknown"
|
||
announced = "announces prefixes" if data.get("announced") else "announces no prefixes"
|
||
return (f"## Routing (BGP)\nAS{asn} holder: {holder}; {announced} in the current routing "
|
||
f"view.\nFrom RIPEstat, {url}")
|
||
|
||
|
||
def _format_rdap(target: str, kind: str, data: dict) -> str:
|
||
lines = [f"# RDAP registration for {target} ({kind})", ""]
|
||
|
||
if name := data.get("name"):
|
||
lines.append(f"Name: {name}")
|
||
if handle := data.get("handle"):
|
||
lines.append(f"Handle: {handle}")
|
||
if start := data.get("startAddress"):
|
||
lines.append(f"Range: {start} - {data.get('endAddress', '?')}")
|
||
if cidrs := data.get("cidr0_cidrs"):
|
||
rendered = ", ".join(
|
||
f"{c.get('v4prefix') or c.get('v6prefix')}/{c.get('length')}" for c in cidrs[:4]
|
||
)
|
||
lines.append(f"CIDR: {rendered}")
|
||
if country := data.get("country"):
|
||
lines.append(f"Country: {country}")
|
||
if atype := data.get("type"):
|
||
lines.append(f"Type: {atype}")
|
||
|
||
for event in data.get("events") or []:
|
||
action = event.get("eventAction", "")
|
||
when = (event.get("eventDate") or "")[:10]
|
||
if action in ("registration", "last changed", "expiration"):
|
||
lines.append(f"{action.title():10} {when}")
|
||
|
||
if statuses := data.get("status"):
|
||
lines.append(f"Status: {', '.join(statuses[:6])}")
|
||
|
||
contacts: list[str] = []
|
||
for entity in data.get("entities") or []:
|
||
roles = ", ".join(entity.get("roles") or [])
|
||
name, email = _vcard(entity.get("vcardArray"))
|
||
if name or email:
|
||
contacts.append(f"- {roles or 'contact'}: {name}{f' <{email}>' if email else ''}")
|
||
if contacts:
|
||
lines += ["", "## Contacts"] + contacts[:6]
|
||
|
||
if nameservers := data.get("nameservers"):
|
||
lines += ["", "## Nameservers"] + [f"- {ns.get('ldhName', '')}" for ns in nameservers[:8]]
|
||
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _vcard(vcard) -> tuple[str, str]:
|
||
name = email = ""
|
||
if not isinstance(vcard, list) or len(vcard) < 2:
|
||
return name, email
|
||
for entry in vcard[1]:
|
||
if not isinstance(entry, list) or len(entry) < 4:
|
||
continue
|
||
if entry[0] == "fn":
|
||
name = str(entry[3])
|
||
elif entry[0] == "email":
|
||
email = str(entry[3])
|
||
return name, email
|
||
|
||
|
||
# ---------------------------------------------------- network_reference (offline)
|
||
|
||
class NetworkRefRequest(ToolModel):
|
||
lookup_type: Literal["mac_vendor", "port", "http_status"] = Field(
|
||
description="mac_vendor=manufacturer of a MAC address; port=service registered on a "
|
||
"TCP/UDP port; http_status=meaning of an HTTP status code.",
|
||
)
|
||
value: str = Field(
|
||
description="A MAC address ('00:1A:2B:3C:4D:5E'), a port number ('443'), or an HTTP status "
|
||
"code ('418').",
|
||
)
|
||
|
||
|
||
@router.post("/network_reference", operation_id="network_reference", response_model=str,
|
||
summary="Look up MAC vendor, port or HTTP status")
|
||
@tool
|
||
async def network_reference(body: NetworkRefRequest) -> str:
|
||
"""Look up a MAC address vendor, a TCP/UDP port's registered service, or an HTTP status code.
|
||
|
||
Answers from the IEEE OUI and IANA port registries held locally. Use it rather than
|
||
recalling port numbers or OUI prefixes.
|
||
"""
|
||
value = body.value.strip()
|
||
if not value:
|
||
return "CANNOT DO THAT: provide a value to look up."
|
||
|
||
if body.lookup_type == "mac_vendor":
|
||
return _lookup_mac(value)
|
||
if body.lookup_type == "port":
|
||
return _lookup_port(value)
|
||
return _lookup_http_status(value)
|
||
|
||
|
||
@lru_cache(maxsize=1)
|
||
def _oui_table() -> dict[str, str]:
|
||
"""Load Wireshark's manuf file.
|
||
|
||
Preferred over IEEE's oui.csv because it already merges the MA-L, MA-M
|
||
and MA-S registries with explicit prefix lengths. Loading only the 24-bit
|
||
registry silently mis-attributes every address in a 28- or 36-bit block
|
||
to whoever owns the enclosing /24.
|
||
"""
|
||
table: dict[str, str] = {}
|
||
path = DATA_DIR / "manuf"
|
||
if not path.exists():
|
||
return table
|
||
|
||
with path.open(encoding="utf-8", errors="replace") as handle:
|
||
for line in handle:
|
||
if line.startswith("#") or not line.strip():
|
||
continue
|
||
fields = line.rstrip("\n").split("\t")
|
||
if len(fields) < 2:
|
||
continue
|
||
prefix = fields[0].strip().upper()
|
||
vendor = (fields[2] if len(fields) > 2 and fields[2].strip() else fields[1]).strip()
|
||
table[prefix] = vendor
|
||
return table
|
||
|
||
|
||
def _lookup_mac(value: str) -> str:
|
||
hex_digits = re.sub(r"[^0-9A-Fa-f]", "", value).upper()
|
||
if len(hex_digits) < 6:
|
||
return f"CANNOT DO THAT: '{value}' is not a recognisable MAC address."
|
||
|
||
table = _oui_table()
|
||
if not table:
|
||
return ("The OUI database is not available in this build. Do not guess the vendor "
|
||
"from the prefix.")
|
||
|
||
normalised = ":".join(hex_digits[i:i + 2] for i in range(0, min(len(hex_digits), 12), 2))
|
||
|
||
# Longest-prefix match: 36-bit, then 28-bit, then 24-bit blocks.
|
||
for bits in (36, 28, 24):
|
||
nibbles = bits // 4
|
||
if len(hex_digits) < nibbles:
|
||
continue
|
||
candidate = hex_digits[:nibbles]
|
||
octets = ":".join(candidate[i:i + 2] for i in range(0, len(candidate) - len(candidate) % 2, 2))
|
||
for key in (f"{octets}/{bits}", octets):
|
||
if vendor := table.get(key.upper()):
|
||
out = [f"# {normalised}", "", f"**Vendor: {vendor}**", f"Matched OUI block: {key} ({bits}-bit)"]
|
||
out.append(_mac_flags(hex_digits))
|
||
return "\n".join(out)
|
||
|
||
return (f"# {normalised}\n\nNo vendor is registered for this OUI prefix.\n"
|
||
f"{_mac_flags(hex_digits)}\n\nDo not guess the manufacturer.")
|
||
|
||
|
||
def _mac_flags(hex_digits: str) -> str:
|
||
first = int(hex_digits[0:2], 16)
|
||
notes = []
|
||
if first & 0x01:
|
||
notes.append("multicast/group address")
|
||
else:
|
||
notes.append("unicast")
|
||
if first & 0x02:
|
||
notes.append("**locally administered** — this is a randomised or manually set address, "
|
||
"so the vendor prefix does not identify the real hardware")
|
||
else:
|
||
notes.append("globally unique (burned-in)")
|
||
return "\nFlags: " + "; ".join(notes)
|
||
|
||
|
||
@lru_cache(maxsize=1)
|
||
def _port_table() -> dict[tuple[str, str], list[tuple[str, str]]]:
|
||
table: dict[tuple[str, str], list[tuple[str, str]]] = {}
|
||
path = DATA_DIR / "iana-ports.csv"
|
||
if not path.exists():
|
||
return table
|
||
|
||
with path.open(encoding="utf-8", errors="replace", newline="") as handle:
|
||
for row in csv.DictReader(handle):
|
||
port = (row.get("Port Number") or "").strip()
|
||
proto = (row.get("Transport Protocol") or "").strip().lower()
|
||
service = (row.get("Service Name") or "").strip()
|
||
desc = (row.get("Description") or "").strip()
|
||
if not port or not proto or not service or "-" in port:
|
||
continue
|
||
table.setdefault((port, proto), []).append((service, desc))
|
||
return table
|
||
|
||
|
||
def _lookup_port(value: str) -> str:
|
||
match = re.search(r"\d+", value)
|
||
if not match:
|
||
return f"CANNOT DO THAT: '{value}' does not contain a port number."
|
||
port = match.group()
|
||
if not 0 <= int(port) <= 65535:
|
||
return f"CANNOT DO THAT: {port} is outside the valid port range 0-65535."
|
||
|
||
table = _port_table()
|
||
if not table:
|
||
return "The IANA port registry is not available in this build."
|
||
|
||
lines = [f"# Port {port}", ""]
|
||
found = False
|
||
for proto in ("tcp", "udp", "sctp", "dccp"):
|
||
for service, desc in table.get((port, proto), []):
|
||
found = True
|
||
lines.append(f"- **{proto.upper()}/{port}** — `{service}`{f': {desc}' if desc else ''}")
|
||
|
||
if not found:
|
||
lines.append("No service is registered with IANA on this port. It may still be used by "
|
||
"an application unofficially, but there is no assignment to cite.")
|
||
|
||
number = int(port)
|
||
if number < 1024:
|
||
lines += ["", "This is a well-known port (0-1023); binding it requires root on Unix."]
|
||
elif number < 49152:
|
||
lines += ["", "This is a registered port (1024-49151)."]
|
||
else:
|
||
lines += ["", "This is in the dynamic/ephemeral range (49152-65535)."]
|
||
|
||
return "\n".join(lines)
|
||
|
||
|
||
_HTTP_STATUS = {
|
||
100: "Continue", 101: "Switching Protocols", 103: "Early Hints",
|
||
200: "OK", 201: "Created", 202: "Accepted", 204: "No Content", 206: "Partial Content",
|
||
301: "Moved Permanently", 302: "Found", 303: "See Other", 304: "Not Modified",
|
||
307: "Temporary Redirect", 308: "Permanent Redirect",
|
||
400: "Bad Request", 401: "Unauthorized", 402: "Payment Required", 403: "Forbidden",
|
||
404: "Not Found", 405: "Method Not Allowed", 406: "Not Acceptable",
|
||
407: "Proxy Authentication Required", 408: "Request Timeout", 409: "Conflict",
|
||
410: "Gone", 411: "Length Required", 412: "Precondition Failed",
|
||
413: "Content Too Large", 414: "URI Too Long", 415: "Unsupported Media Type",
|
||
416: "Range Not Satisfiable", 417: "Expectation Failed", 418: "I'm a teapot",
|
||
421: "Misdirected Request", 422: "Unprocessable Content", 425: "Too Early",
|
||
426: "Upgrade Required", 428: "Precondition Required", 429: "Too Many Requests",
|
||
431: "Request Header Fields Too Large", 451: "Unavailable For Legal Reasons",
|
||
500: "Internal Server Error", 501: "Not Implemented", 502: "Bad Gateway",
|
||
503: "Service Unavailable", 504: "Gateway Timeout", 505: "HTTP Version Not Supported",
|
||
507: "Insufficient Storage", 508: "Loop Detected", 511: "Network Authentication Required",
|
||
}
|
||
|
||
_STATUS_NOTES = {
|
||
401: "Means *unauthenticated* despite the name - no or bad credentials. Use 403 for "
|
||
"authenticated-but-not-allowed.",
|
||
403: "Authenticated but not permitted. If the user is getting this from a proxy, the "
|
||
"proxy may be blocking the destination rather than the origin refusing.",
|
||
404: "The server has no representation for this URI. It does not distinguish 'never "
|
||
"existed' from 'deleted' - use 410 for the latter.",
|
||
429: "Rate limited. Check for a Retry-After header before retrying.",
|
||
502: "An upstream server returned something invalid to a gateway or proxy. The fault is "
|
||
"usually behind the proxy, not in it.",
|
||
504: "A gateway timed out waiting for an upstream. Distinguish from 502: nothing came back at all.",
|
||
}
|
||
|
||
|
||
def _lookup_http_status(value: str) -> str:
|
||
match = re.search(r"\d{3}", value)
|
||
if not match:
|
||
return f"CANNOT DO THAT: '{value}' does not contain a three-digit HTTP status code."
|
||
|
||
code = int(match.group())
|
||
name = _HTTP_STATUS.get(code)
|
||
if not name:
|
||
klass = {1: "informational", 2: "success", 3: "redirection",
|
||
4: "client error", 5: "server error"}.get(code // 100)
|
||
if not klass:
|
||
return f"{code} is not a valid HTTP status code."
|
||
return f"# HTTP {code}\n\nNot a standard status code, but it is in the {klass} class ({code // 100}xx)."
|
||
|
||
klass = {1: "Informational", 2: "Success", 3: "Redirection",
|
||
4: "Client error", 5: "Server error"}[code // 100]
|
||
lines = [f"# HTTP {code} {name}", "", f"Class: {klass} ({code // 100}xx)"]
|
||
if note := _STATUS_NOTES.get(code):
|
||
lines += ["", note]
|
||
return "\n".join(lines)
|