grocery-manager/app/rest/validate.py

#
# Copyright (c) Daniel Sheffield 2023
# All rights reserved
#
# THIS SOFTWARE IS PROVIDED AS IS WITHOUT WARRANTY

# https://www.ietf.org/rfc/rfc3696.txt

"""
   Without quotes, local-parts may consist of any combination of
   alphabetic characters, digits, or any of the special characters

      ! # $ % & ' * + - / = ?  ^ _ ` . { | } ~

   period (".") may also appear, but may not be used to start or end the
   local part, nor may two or more consecutive periods appear.  Stated
   differently, any ASCII graphic (printing) character other than the
   at-sign ("@"), backslash, double quote, comma, or square brackets may
   appear without quoting.  If any of that list of excluded characters
   are to appear, they must be quoted.
"""

from io import BufferedReader
from itertools import chain, zip_longest
from bottle import static_file, HTTPError, abort, LocalRequest
from urllib.parse import urlparse, quote, quote_plus, quote_from_bytes, urlencode
from .hash_util import bytes_to_base32, blake

# according to rfc3696
URL_MUST_ESCAPE = bytes([
    x for x in chain(
        # control characters
        range(int('0x1F', 0)+1),
        # 0x7F and non 7bit-ASCII
        range(int('0x7F', 0,), int('0xFF', 0)+1),
        # specifically excluded
        b'@\\",[]'
    )
])
# so give this list to urllib.parse.quote which follows rfc3986
URL_SAFE = bytes(( i for i in range(int('0xff',0)+1) if i not in map(int, URL_MUST_ESCAPE) ))

CLIP_SIZE_LIMIT = 65535
def validate(filename: str) -> bytes:
    ret = static_file('/'.join([filename,]*2) + '.file', root='app/rest/static')
    if isinstance(ret, HTTPError):
        return abort(404, f"No such paste: {filename}")

    if ret.content_length > CLIP_SIZE_LIMIT:
        return abort(418, f"Paste size exceeds {CLIP_SIZE_LIMIT}")

    content: bytes = ret.body.read() if isinstance(ret.body, BufferedReader) else ret.body.encode('utf-8')

    _bytes = blake(content, person='clip'.encode('utf-8'))
    _b32 = bytes_to_base32(_bytes)
    if _b32 != filename:
        return abort(410, f"Paste content differs")
    return content


def validate_parameter(request: LocalRequest, name: str) -> bytes:
    if name not in request.params:
        return abort(400, f"Missing parameter: '{name}'")
    
    # TODO: what is correct overhead for form content?
    OVERHEAD = 1024
    content: bytes = request.query.get(name, None)
    content_length = request.content_length
    if content_length == -1:
        return abort(418, f"Content-Length must be specified")
    if content_length > CLIP_SIZE_LIMIT + OVERHEAD:
        return abort(418, f"Content-Length can not exceed {CLIP_SIZE_LIMIT*3} bytes")

    # TODO: add test for both query/form param
    if 'multipart/form-data' in request.content_type:
        # TODO: what about binary data ?
        content: bytes = (content or request.params[name].encode('utf-8'))
    else:
        content: bytes = (content or request.params[name].encode('latin-1'))
    
    if len(content) > CLIP_SIZE_LIMIT:
        return abort(418, f"Paste can not exceed {CLIP_SIZE_LIMIT} bytes")
    return content

def validate_url(url: str) -> str:
    scheme, netloc, path, params, query, fragment = urlparse(url)

    if not scheme: return abort(400, "URL has no scheme")

    if scheme == 'file' and not path: return abort(400, "File URL has no path")

    if scheme in ('http', 'https') and not netloc: return abort(400, "HTTP(S) URL has no netloc")

    if netloc:
        try:
            user_info, loc = netloc.rsplit('@', 1)
        except ValueError:
            user_info = ''
            loc = ''
        if user_info:
            user_info = quote(user_info, safe=URL_SAFE)
            netloc = f"{user_info}@{''.join(loc)}"
        else:
            # TODO: do this properly, ie, valid dns-name/ip/port etc
            netloc = quote(netloc, safe=URL_SAFE)
    
    path = quote(path, safe=URL_SAFE)
    params = quote_plus(params, safe=URL_SAFE)
    query = quote(query, safe=URL_SAFE)
    fragment = quote(fragment, safe=URL_SAFE)
    
    url = f'{scheme}://{netloc}{path}{params}'
    if query:
        url = f'{url}?{query}'
    if fragment:
        url = f'{url}#{fragment}'
    return url