F5 iRule Linter

#!/usr/bin/env python3
"""
F5 iRule Linter

READ THE DOCUMENTATION BEFORE YOU RUN THIS:
    https://permitany.com/scripts/f5-irule-linter

It covers what the script does, whether it is safe to run, the
inventory file and credentials it needs, how to read the output,
and what to do when something fails.

    # No extra packages — uses Python stdlib only
"""

import argparse
import json
import os
import re
import sys

sys.stdout.reconfigure(encoding="utf-8", errors="replace")

SEV_ORDER = {"CRITICAL": 0, "HIGH": 1, "MEDIUM": 2, "LOW": 3}

# Commands that are only valid once a given parse stage has happened. An iRule
# using one in an earlier event saves cleanly and throws at runtime, on live
# traffic, which is why this is worth catching offline.
EVENT_REQUIRES = {
    "HTTP::": ("HTTP_REQUEST", "HTTP_REQUEST_DATA", "HTTP_REQUEST_SEND",
               "HTTP_RESPONSE", "HTTP_RESPONSE_DATA", "HTTP_RESPONSE_RELEASE",
               "LB_SELECTED", "LB_FAILED", "SERVER_CONNECTED"),
    "SSL::": ("CLIENTSSL_HANDSHAKE", "CLIENTSSL_CLIENTHELLO", "CLIENTSSL_SERVERHELLO_SEND",
              "SERVERSSL_HANDSHAKE", "HTTP_REQUEST", "HTTP_RESPONSE"),
    "LB::": ("LB_SELECTED", "LB_FAILED", "SERVER_CONNECTED", "HTTP_RESPONSE",
             "HTTP_RESPONSE_DATA", "SERVER_CLOSED"),
}

DATA_EVENTS = ("HTTP_REQUEST_DATA", "HTTP_RESPONSE_DATA")
TCP_DATA_EVENTS = ("CLIENT_DATA", "SERVER_DATA")

# Events on the hot path: anything here runs per request or per connection.
HOT_EVENTS = ("HTTP_REQUEST", "HTTP_RESPONSE", "HTTP_REQUEST_DATA",
              "HTTP_RESPONSE_DATA", "CLIENT_ACCEPTED", "CLIENT_DATA",
              "SERVER_DATA", "LB_SELECTED")


class Finding:
    def __init__(self, severity, code, line, event, message, fix):
        self.severity = severity
        self.code = code
        self.line = line
        self.event = event
        self.message = message
        self.fix = fix

    def as_dict(self):
        return {
            "severity": self.severity, "code": self.code, "line": self.line,
            "event": self.event, "message": self.message, "fix": self.fix,
        }


def strip_comments(line):
    """Remove a trailing TCL comment, leaving string literals alone.

    Crude on purpose: a '#' inside a quoted string is kept, everything from an
    unquoted '#' is dropped. Good enough to stop comments producing findings,
    which is the only thing this is for.
    """
    out, in_str, i = [], False, 0
    while i < len(line):
        c = line[i]
        if c == "\\" and i + 1 < len(line):
            out.append(line[i:i + 2])
            i += 2
            continue
        if c == '"':
            in_str = not in_str
        if c == "#" and not in_str:
            break
        out.append(c)
        i += 1
    return "".join(out)


def parse_events(lines):
    """Split the file into (event_name, start_line, end_line) blocks.

    Brace counting rather than a regex, because an iRule nests braces several
    deep and a regex that matches to the first closing brace cuts every block
    short - the same failure the template engine had.
    """
    events, i, n = [], 0, len(lines)
    while i < n:
        m = re.match(r"\s*when\s+([A-Z_0-9]+)\s*\{", strip_comments(lines[i]))
        if not m:
            i += 1
            continue
        name, start, depth = m.group(1), i, 0
        for j in range(i, n):
            code = strip_comments(lines[j])
            depth += code.count("{") - code.count("}")
            if depth <= 0 and j > i:
                events.append((name, start, j))
                i = j + 1
                break
            if depth <= 0 and j == i and code.count("}") >= code.count("{"):
                events.append((name, start, j))
                i = j + 1
                break
        else:
            events.append((name, start, n - 1))
            i = n
    return events


def check_braces(lines, findings):
    depth = 0
    for idx, raw in enumerate(lines, 1):
        code = strip_comments(raw)
        depth += code.count("{") - code.count("}")
        if depth < 0:
            findings.append(Finding(
                "CRITICAL", "SYNTAX-BRACE", idx, "-",
                "More closing braces than opening ones at this line.",
                "The iRule will not load. Balance the braces before anything else."))
            return
    if depth != 0:
        findings.append(Finding(
            "CRITICAL", "SYNTAX-BRACE", len(lines), "-",
            "Unbalanced braces: %d brace(s) never closed." % depth,
            "The iRule will not load. Balance the braces before anything else."))


def check_release(lines, event, start, end, findings):
    """Every path out of a *_DATA event must call HTTP::release / TCP::release.

    This is the check that matters most: a missed release on an early return
    hangs the connection until it times out, and it is the most common way an
    iRule takes a service down.
    """
    body = [strip_comments(l) for l in lines[start:end + 1]]
    joined = "\n".join(body)
    is_http = event in DATA_EVENTS
    is_tcp = event in TCP_DATA_EVENTS
    if not (is_http or is_tcp):
        return
    rel = "HTTP::release" if is_http else "TCP::release"

    if rel not in joined:
        findings.append(Finding(
            "CRITICAL", "MISSING-RELEASE", start + 1, event,
            "%s is never called in this event." % rel,
            "Add %s on every path out of the event. Without it the connection "
            "hangs until it times out." % rel))
        return

    # A `return` that is not preceded by a release on the same path.
    for off, code in enumerate(body):
        if re.search(r"(^|\s)return(\s|$)", code):
            before = "\n".join(body[:off + 1])
            if rel not in before:
                findings.append(Finding(
                    "CRITICAL", "MISSING-RELEASE", start + off + 1, event,
                    "This `return` leaves the event without calling %s first." % rel,
                    "Call %s before every early return, or restructure so the "
                    "release is unavoidable." % rel))


def check_event_scope(event, start, body, findings):
    for prefix, valid in EVENT_REQUIRES.items():
        if event in valid:
            continue
        for off, code in enumerate(body):
            for m in re.finditer(re.escape(prefix) + r"[a-zA-Z_]+", code):
                findings.append(Finding(
                    "HIGH", "WRONG-EVENT", start + off + 1, event,
                    "`%s` used in %s, where it is not available." % (m.group(0), event),
                    "Move it to one of: %s. Using a command in the wrong event is a "
                    "RUNTIME error, so the iRule saves cleanly and fails on live "
                    "traffic." % ", ".join(valid[:4])))


def check_performance(event, start, body, findings):
    hot = event in HOT_EVENTS
    for off, code in enumerate(body):
        line_no = start + off + 1

        if re.search(r"\b(regexp|regsub)\b", code) and hot:
            findings.append(Finding(
                "MEDIUM", "REGEX-HOT-PATH", line_no, event,
                "Regular expression evaluated on every request.",
                "Prefer `string match`, `starts_with`, `contains` or `switch -glob`. "
                "The regex engine is the most expensive construct available here."))

        if re.search(r"(^|[{;])\s*log\s+", code) and hot:
            if not re.search(r"static::\w*(debug|log)", "\n".join(body[max(0, off - 3):off + 1])):
                findings.append(Finding(
                    "MEDIUM", "UNGUARDED-LOG", line_no, event,
                    "Unguarded `log` on the data path.",
                    "/var/log/ltm is a SYNCHRONOUS disk write. Guard it behind a "
                    "`static::debug` flag set in RULE_INIT, or send it with HSL::send."))

        if re.search(r"\bmatchclass\b", code):
            findings.append(Finding(
                "MEDIUM", "DEPRECATED", line_no, event,
                "`matchclass` is deprecated.",
                "Use `class match`. It is faster and `matchclass` is not guaranteed "
                "to survive an upgrade."))

        if re.search(r"(^|[{;\s])node\s+[\d$\[]", code):
            findings.append(Finding(
                "HIGH", "NODE-BYPASSES-POOL", line_no, event,
                "`node` bypasses the pool, its monitors and persistence.",
                "Traffic goes to that address whether it is healthy or not. Use "
                "`pool <name>` or `pool <name> member <addr> <port>` unless going "
                "out of band is deliberate."))

        if re.search(r"\bHTTP::redirect\b", code):
            findings.append(Finding(
                "LOW", "REDIRECT-302", line_no, event,
                "`HTTP::redirect` sends a 302 (temporary).",
                "For a permanent redirect use `HTTP::respond 301` with an explicit "
                "Location header. A 302 does not pass search ranking."))

        if re.search(r"\bHTTP::collect\s*$", code) or re.search(r"\bHTTP::collect\s*\}", code):
            findings.append(Finding(
                "HIGH", "UNBOUNDED-COLLECT", line_no, event,
                "`HTTP::collect` with no length collects everything available.",
                "Pass a bounded length, e.g. the Content-Length, so a large upload "
                "does not sit in memory."))


def check_cmp(event, start, body, findings):
    for off, code in enumerate(body):
        line_no = start + off + 1
        if re.search(r"\b(table|session)\s+(set|incr|lookup|delete|add|replace|keys|timeout|lifetime)\b", code):
            findings.append(Finding(
                "HIGH", "CMP-RISK", line_no, event,
                "Shared state (`%s`) can demote the virtual server out of CMP."
                % re.search(r"\b(table|session)\b", code).group(1),
                "Verify after deploying: tmsh show ltm virtual VS | grep -i cmp. "
                "Demotion cuts throughput to a single TMM and logs NOTHING."))
        m = re.match(r"\s*set\s+(::\w+)", code)
        if m:
            findings.append(Finding(
                "HIGH", "CMP-RISK", line_no, event,
                "Global variable `%s` is shared state and risks CMP demotion." % m.group(1),
                "Use a `static::` variable set in RULE_INIT for read-only data, or "
                "the `table` command with the CMP caveat understood."))


def check_rule_init(events, lines, findings):
    """Constants computed per request that belong in RULE_INIT."""
    has_init = any(e[0] == "RULE_INIT" for e in events)
    for name, start, end in events:
        if name not in HOT_EVENTS:
            continue
        for off in range(start, end + 1):
            code = strip_comments(lines[off])
            m = re.match(r'\s*set\s+(\w+)\s+"[^"$\[]{40,}"\s*$', code)
            if m:
                findings.append(Finding(
                    "LOW", "CONSTANT-PER-REQUEST", off + 1, name,
                    "Constant string assigned on every request.",
                    "Move it to RULE_INIT as `static::%s`. RULE_INIT runs once at "
                    "load; this runs forever." % m.group(1)
                    + ("" if has_init else " The iRule has no RULE_INIT block yet.")))


def check_if_chain(event, start, body, findings):
    run, run_start = 0, 0
    for off, code in enumerate(body):
        if re.search(r"^\s*\}?\s*(else)?if\s*\{", code):
            if run == 0:
                run_start = off
            run += 1
        elif code.strip() and not code.strip().startswith("}"):
            if run >= 4:
                findings.append(Finding(
                    "LOW", "IF-CHAIN", start + run_start + 1, event,
                    "Chain of %d conditionals evaluated top to bottom per request." % run,
                    "Move the table to a data group and use `class match`. One indexed "
                    "lookup instead of N comparisons, and an operator can edit it "
                    "without touching data-path code."))
            run = 0


def lint(path):
    with open(path, "r", encoding="utf-8", errors="replace") as fh:
        lines = fh.read().replace("\r\n", "\n").split("\n")

    findings = []
    check_braces(lines, findings)
    events = parse_events(lines)

    if not events:
        findings.append(Finding(
            "CRITICAL", "NO-EVENTS", 1, "-",
            "No `when <EVENT> {` block found.",
            "An iRule is a set of event handlers. A file with none does nothing."))
        return findings, events

    for name, start, end in events:
        body = [strip_comments(l) for l in lines[start:end + 1]]
        check_release(lines, name, start, end, findings)
        check_event_scope(name, start, body, findings)
        check_performance(name, start, body, findings)
        check_cmp(name, start, body, findings)
        check_if_chain(name, start, body, findings)

    check_rule_init(events, lines, findings)
    findings.sort(key=lambda f: (SEV_ORDER[f.severity], f.line))
    return findings, events


def report_text(path, findings, events):
    print("=" * 72)
    print("irule-linter  %s" % path)
    print("=" * 72)
    print("Events: %s" % (", ".join("%s (line %d)" % (n, s + 1) for n, s, _ in events) or "none"))
    print()
    if not findings:
        print("No findings. Note this is STATIC analysis: it cannot tell you whether")
        print("an LTM policy would do the same job compiled, which is still the first")
        print("question to ask.")
        return
    counts = {}
    for f in findings:
        counts[f.severity] = counts.get(f.severity, 0) + 1
    print("Findings: " + "  ".join("%s=%d" % (s, counts[s])
                                   for s in ("CRITICAL", "HIGH", "MEDIUM", "LOW")
                                   if s in counts))
    print()
    for f in findings:
        print("[%s] %s  line %d  (%s)" % (f.severity, f.code, f.line, f.event))
        print("    %s" % f.message)
        print("    FIX: %s" % f.fix)
        print()


def main():
    ap = argparse.ArgumentParser(
        description="Static analysis for F5 iRules. Reads a .tcl file; never "
                    "connects to a BIG-IP.")
    ap.add_argument("files", nargs="+", help="One or more iRule files (.tcl)")
    ap.add_argument("--json", action="store_true", help="Machine-readable output")
    ap.add_argument("--fail-on", default="HIGH",
                    choices=["CRITICAL", "HIGH", "MEDIUM", "LOW"],
                    help="Lowest severity that sets exit code 1 (default: HIGH)")
    args = ap.parse_args()

    all_results, worst = [], None
    for path in args.files:
        if not os.path.isfile(path):
            print("ERROR: not a file: %s" % path, file=sys.stderr)
            return 2
        try:
            findings, events = lint(path)
        except (OSError, UnicodeError) as exc:
            print("ERROR: could not read %s: %s" % (path, exc), file=sys.stderr)
            return 2
        all_results.append({
            "file": path,
            "events": [{"event": n, "line": s + 1} for n, s, _ in events],
            "findings": [f.as_dict() for f in findings],
        })
        for f in findings:
            if worst is None or SEV_ORDER[f.severity] < SEV_ORDER[worst]:
                worst = f.severity
        if not args.json:
            report_text(path, findings, events)

    if args.json:
        print(json.dumps({"results": all_results}, indent=2))

    if worst is None:
        return 0
    if worst == "CRITICAL":
        return 2
    return 1 if SEV_ORDER[worst] <= SEV_ORDER[args.fail_on] else 0


if __name__ == "__main__":
    sys.exit(main())