monitor.parse-access-log

Imports name this capability’s declared dependencies, which fune builds next to it in your project; each one links to its page.

from typing import Optional

from .time_iso_to_unix import iso_to_unix  ← from time.iso-to-unix ^1.0.0 · built alongside by fune
from .monitor_parse_access_log_types import AccessLogEntry

_MONTHS = ["Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"]


def _at(t: str, i: int) -> str:
    return t[i] if 0 <= i < len(t) else ""


def _digits(text: str, start: int, count: int) -> Optional[int]:
    n = 0
    for i in range(start, start + count):
        c = _at(text, i)
        # Not str.isdigit(): that accepts other scripts' digits and superscripts.
        if not ("0" <= c <= "9") or c == "":
            return None
        n = n * 10 + ord(c) - 48
    return n


def _dash(field: str) -> Optional[str]:
    return None if field == "-" else field


def _parse_time(t: str, i: int) -> Optional[int]:
    """The time field, e.g. [10/Oct/2000:13:55:36 -0700], starting at `i` to Unix seconds, or None."""
    for off, ch in ((0, "["), (3, "/"), (7, "/"), (12, ":"), (15, ":"), (18, ":"), (21, " "), (27, "]")):
        if _at(t, i + off) != ch:
            return None
    day = _digits(t, i + 1, 2)
    name = t[i + 4:i + 7]
    month = _MONTHS.index(name) + 1 if name in _MONTHS else 0
    year = _digits(t, i + 8, 4)
    hour = _digits(t, i + 13, 2)
    minute = _digits(t, i + 16, 2)
    second = _digits(t, i + 19, 2)
    sign = _at(t, i + 22)
    oh = _digits(t, i + 23, 2)
    om = _digits(t, i + 25, 2)
    if day is None or year is None or hour is None or minute is None or second is None or oh is None or om is None:
        return None
    if month == 0 or year < 1 or hour > 23 or minute > 59 or second > 59 or oh > 23 or om > 59:
        return None
    if sign != "+" and sign != "-":
        return None
    leap = (year % 4 == 0 and year % 100 != 0) or year % 400 == 0
    month_days = [31, 29 if leap else 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31][month - 1]
    if day < 1 or day > month_days:
        return None
    # Every field is checked above, so iso_to_unix cannot raise here.
    return iso_to_unix("%s-%02d-%sT%s%s%s:%s" % (t[i + 8:i + 12], month, t[i + 1:i + 3], t[i + 13:i + 21], sign, t[i + 23:i + 25], t[i + 25:i + 27]))


def _closing_quote(t: str, i: int) -> int:
    """The index of the quote closing a field that opens at `i`, honouring \\" escapes; -1 if none."""
    if _at(t, i) != '"':
        return -1
    j = i + 1
    while j < len(t):
        if t[j] == "\\":
            j += 1
        elif t[j] == '"':
            return j
        j += 1
    return -1


def _is_method(m: str) -> bool:
    return len(m) > 0 and all("A" <= c <= "Z" for c in m)


def parse_access_log(line: str) -> Optional[AccessLogEntry]:
    """One Common or Combined Log Format line, or None when the line is in
    neither format: logs hold junk, and a reader should skip it rather than stop."""
    t = line
    while t.endswith("\n") or t.endswith("\r"):
        t = t[:-1]
    host_end = t.find(" ")
    if host_end < 1:
        return None
    ident_end = t.find(" ", host_end + 1)
    if ident_end < host_end + 2:
        return None
    user_end = t.find(" [", ident_end + 1)
    if user_end < ident_end + 2:
        return None
    at = _parse_time(t, user_end + 1)
    if at is None:
        return None
    i = user_end + 29
    if _at(t, i) != " ":
        return None
    req_end = _closing_quote(t, i + 1)
    if req_end < 0:
        return None
    request = t[i + 2:req_end]
    i = req_end + 1
    # A three-digit status, then the byte count: both are required.
    if _at(t, i) != " " or _at(t, i + 4) != " ":
        return None
    status = _digits(t, i + 1, 3)
    if status is None or status < 100 or status > 599:
        return None
    i += 4
    bytes_end = t.find(" ", i + 1)
    if bytes_end < 0:
        bytes_end = len(t)
    bytes_text = t[i + 1:bytes_end]
    size: Optional[int] = None
    if bytes_text != "-":
        if len(bytes_text) < 1 or len(bytes_text) > 15:
            return None
        size = _digits(bytes_text, 0, len(bytes_text))
        if size is None:
            return None
    referer: Optional[str] = None
    user_agent: Optional[str] = None
    if bytes_end < len(t):
        ref_end = _closing_quote(t, bytes_end + 1)
        if ref_end < 0 or _at(t, ref_end + 1) != " ":
            return None
        ua_end = _closing_quote(t, ref_end + 2)
        if ua_end < 0 or (ua_end + 1 < len(t) and t[ua_end + 1] != " "):
            return None
        referer = _dash(t[bytes_end + 2:ref_end])
        user_agent = _dash(t[ref_end + 3:ua_end])
    method: Optional[str] = None
    path: Optional[str] = None
    protocol: Optional[str] = None
    parts = request.split(" ")
    if len(parts) == 3 and _is_method(parts[0]) and parts[1] != "" and parts[2].startswith("HTTP/"):
        method, path, protocol = parts
    elif len(parts) == 2 and _is_method(parts[0]) and parts[1] != "":
        method, path = parts
    return AccessLogEntry(
        remote_host=t[:host_end],
        ident=_dash(t[host_end + 1:ident_end]),
        user=_dash(t[ident_end + 1:user_end]),
        at=at,
        method=method,
        path=path,
        protocol=protocol,
        status=status,
        bytes=size,
        referer=referer,
        user_agent=user_agent,
    )