from typing import Optional from .time_iso_to_unix import iso_to_unix 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, )