cms-SiteForge/app/safe_http.py

"""Bounded outbound HTTP with DNS-to-socket pinning.

Only public HTTP(S) hosts are permitted. A resolved address is pinned to
its socket while TLS verification still uses the original hostname; this
prevents a second, unchecked DNS lookup between URL validation and connect.
No proxy, cookie jar, or automatic redirect handling is used.
"""
from __future__ import annotations
import http.client
import ipaddress
import json
import os
import socket
import ssl
from dataclasses import dataclass
from urllib.parse import urljoin, urlsplit, urlunsplit

class OutboundError(ValueError):
    pass

@dataclass(frozen=True)
class Fetched:
    url: str
    status_code: int
    headers: dict
    content: bytes
    @property
    def text(self):
        charset='utf-8'
        ct=self.headers.get('content-type','')
        if 'charset=' in ct:charset=ct.split('charset=',1)[1].split(';',1)[0].strip()
        try:return self.content.decode(charset,errors='replace')
        except LookupError:return self.content.decode('utf-8',errors='replace')
    def json(self):return json.loads(self.content)
    @property
    def is_redirect(self):return self.status_code in (301,302,303,307,308)
    def raise_for_status(self):
        if self.status_code>=400:raise OutboundError(f'Upstream HTTP {self.status_code}')

class _PinnedHTTP(http.client.HTTPConnection):
    def __init__(self,hostname,ip,port,timeout):
        self._fixed_ip=ip
        super().__init__(hostname,port=port,timeout=timeout)
    def connect(self):
        self.sock=socket.create_connection((self._fixed_ip,self.port),self.timeout)

class _PinnedHTTPS(http.client.HTTPSConnection):
    def __init__(self,hostname,ip,port,timeout):
        self._fixed_ip=ip
        super().__init__(hostname,port=port,timeout=timeout,context=ssl.create_default_context())
    def connect(self):
        plain=socket.create_connection((self._fixed_ip,self.port),self.timeout)
        try:self.sock=self._context.wrap_socket(plain,server_hostname=self.host)
        except BaseException:
            plain.close();raise

def resolve_target(url):
    if not isinstance(url,str) or len(url)>2048 or any(ord(ch)<32 for ch in url):
        raise OutboundError('Invalid outbound URL')
    p=urlsplit(url)
    if p.scheme not in ('http','https') or not p.hostname or p.username is not None or p.password is not None:
        raise OutboundError('Only public HTTP(S) URLs allowed')
    host=p.hostname.lower().rstrip('.')
    if host in ('localhost','localhost.localdomain') or host.endswith(('.local','.internal','.localhost')):
        raise OutboundError('Local network addresses are forbidden')
    try:
        literal=ipaddress.ip_address(host)
    except ValueError:
        literal=None
    if literal is not None and not literal.is_global:
        raise OutboundError('Private or reserved IP literal forbidden')
    try:port=p.port or (443 if p.scheme=='https' else 80)
    except ValueError as ex:raise OutboundError('Invalid port') from ex
    allowed={int(x) for x in os.getenv('SITEFORGE_OUTBOUND_PORTS','80,443').split(',') if x.strip().isdigit()}
    if port not in allowed:raise OutboundError('Outbound port not allowed')
    try:
        infos=socket.getaddrinfo(host,port,type=socket.SOCK_STREAM)
    except (socket.gaierror,UnicodeError,OSError) as ex:raise OutboundError('Host DNS resolution failed') from ex
    ips=list(dict.fromkeys(info[4][0] for info in infos))
    if not ips or any(not ipaddress.ip_address(ip).is_global for ip in ips):
        raise OutboundError('Private, reserved or mixed DNS results forbidden')
    return p,host,port,ips[0]

def fetch(url,*,max_bytes=3_000_000,timeout=12,redirects=0,headers=None,method="GET",body=None,credential_headers=None):
    """Fetch bounded response. Redirect hops are re-resolved and revalidated."""
    if method not in ("GET","POST"):raise OutboundError("Outbound method forbidden")
    if max_bytes<1 or max_bytes>120_000_000:raise OutboundError('Invalid response limit')
    if not (1<=timeout<=120):raise OutboundError('Invalid timeout')
    if not (0<=redirects<=5):raise OutboundError('Invalid redirect limit')
    if credential_headers:
        if not isinstance(credential_headers,dict) or redirects!=0 or urlsplit(url).scheme!='https':
            raise OutboundError('Credentials require HTTPS with redirects disabled')
        for name,value in credential_headers.items():
            low=name.lower()
            if not (low=='authorization' or (low.startswith('x-') and name.replace('-','').isalnum())):
                raise OutboundError('Forbidden credential header')
            if not isinstance(value,str) or not value or len(value)>8192 or any(c in value for c in '\r\n'):
                raise OutboundError('Invalid credential value')
    current=url
    for hop in range(redirects+1):
        p,hostname,port,ip=resolve_target(current)
        conn=(_PinnedHTTPS if p.scheme=='https' else _PinnedHTTP)(hostname,ip,port,timeout)
        path=p.path or '/'
        if p.query:path+='?'+p.query
        try:
            h={'User-Agent':'SiteForgeAuthorizedImport/1.0','Accept':'*/*','Connection':'close'}
            if headers:
                for k,v in headers.items():
                    if k.lower() not in ('host','connection','cookie','authorization','proxy-authorization'):
                        h[k]=v
            if credential_headers:h.update(credential_headers)
            conn.request(method,path,body=body,headers=h)
            resp=conn.getresponse()
            status=resp.status
            hdr={k.lower():v for k,v in resp.getheaders()}
            if status in (301,302,303,307,308) and hop<redirects:
                loc=hdr.get('location')
                if not loc:raise OutboundError('Redirect lacks Location')
                next_url=urljoin(current,loc)
                resolve_target(next_url)  # security gate before any follow-up connection
                current=next_url
                continue
            if status in (301,302,303,307,308):raise OutboundError('Redirect not permitted')
            stated=hdr.get('content-length')
            if stated and stated.isdecimal() and int(stated)>max_bytes:raise OutboundError('Response exceeds size limit')
            raw=resp.read(max_bytes+1)
            if len(raw)>max_bytes:raise OutboundError('Response exceeds size limit')
            return Fetched(current,status,hdr,raw)
        finally:conn.close()
    raise OutboundError('Too many redirects')


def post(url,body=b'',*,headers=None,max_bytes=1_000_000,timeout=120):
    """Pinned HTTPS/HTTP POST, with no redirection or proxy."""
    return fetch(url,headers=headers,max_bytes=max_bytes,timeout=timeout,redirects=0,method='POST',body=body)