path: root/lib9p/protogen/c_validate.py

# lib9p/protogen/c_validate.py - Generate C validation functions
#
# Copyright (C) 2024-2025  Luke T. Shumaker <lukeshu@lukeshu.com>
# SPDX-License-Identifier: AGPL-3.0-or-later

import typing

import idl

from . import c9util, cutil, idlutil

# This strives to be "general-purpose" in that it just acts on the
# *.9p inputs; but (unfortunately?) there are a few special-cases in
# this script, marked with "SPECIAL".


# pylint: disable=unused-variable
__all__ = ["gen_c_validate"]


def should_save_offset(parent: idl.Struct, child: idl.StructMember) -> bool:
    if child.val or child.max or isinstance(child.typ, idl.Bitfield):
        return True
    for sibling in parent.members:
        if sibling.val:
            for tok in sibling.val.tokens:
                if isinstance(tok, idl.ExprSym) and tok.symname == f"&{child.membname}":
                    return True
        if sibling.max:
            for tok in sibling.max.tokens:
                if isinstance(tok, idl.ExprSym) and tok.symname == f"&{child.membname}":
                    return True
    return False


def should_save_end_offset(struct: idl.Struct) -> bool:
    for memb in struct.members:
        if memb.val:
            for tok in memb.val.tokens:
                if isinstance(tok, idl.ExprSym) and tok.symname == "end":
                    return True
        if memb.max:
            for tok in memb.max.tokens:
                if isinstance(tok, idl.ExprSym) and tok.symname == "end":
                    return True
    return False


def gen_c_validate(versions: set[str], typs: list[idl.UserType]) -> str:
    ret = """
/* validate_* *****************************************************************/

"""
    ret += cutil.macro(
        "#define VALIDATE_NET_BYTES(n)\n"
        "\tif (__builtin_add_overflow(net_offset, n, &net_offset))\n"
        "\t\t/* If needed-net-size overflowed uint32_t, then\n"
        "\t\t * there's no way that actual-net-size will live up to\n"
        "\t\t * that.  */\n"
        '\t\treturn lib9p_error(ctx, LINUX_EBADMSG, "message is too short for content");\n'
        "\tif (net_offset > net_size)\n"
        '\t\treturn lib9p_errorf(ctx, LINUX_EBADMSG, "message is too short for content (%"PRIu32" > %"PRIu32") @ %d", net_offset, net_size, __LINE__);\n'
    )
    ret += cutil.macro(
        "#define VALIDATE_NET_UTF8(n)\n"
        "\t{\n"
        "\t\tsize_t len = n;\n"
        "\t\tVALIDATE_NET_BYTES(len);\n"
        "\t\tif (!is_valid_utf8_without_nul(&net_bytes[net_offset-len], len))\n"
        '\t\t\treturn lib9p_error(ctx, LINUX_EBADMSG, "message contains invalid UTF-8");\n'
        "\t}\n"
    )
    ret += cutil.macro(
        "#define RESERVE_HOST_BYTES(n)\n"
        "\tif (__builtin_add_overflow(host_size, n, &host_size))\n"
        "\t\t/* If needed-host-size overflowed ssize_t, then there's\n"
        "\t\t * no way that actual-net-size will live up to\n"
        "\t\t * that.  */\n"
        '\t\treturn lib9p_error(ctx, LINUX_EBADMSG, "message is too short for content");\n'
    )

    ret += "#define GET_U8LE(off)  (net_bytes[off])\n"
    ret += "#define GET_U16LE(off) uint16le_decode(&net_bytes[off])\n"
    ret += "#define GET_U32LE(off) uint32le_decode(&net_bytes[off])\n"
    ret += "#define GET_U64LE(off) uint64le_decode(&net_bytes[off])\n"

    ret += "#define LAST_U8LE()  GET_U8LE(net_offset-1)\n"
    ret += "#define LAST_U16LE() GET_U16LE(net_offset-2)\n"
    ret += "#define LAST_U32LE() GET_U32LE(net_offset-4)\n"
    ret += "#define LAST_U64LE() GET_U64LE(net_offset-8)\n"

    class IndentLevel(typing.NamedTuple):
        ifdef: bool  # whether this is both `{` and `#if`, or just `{`

    indent_stack: list[IndentLevel]

    def ifdef_lvl() -> int:
        return sum(1 if lvl.ifdef else 0 for lvl in indent_stack)

    def indent_lvl() -> int:
        return len(indent_stack)

    incr_buf: int

    def incr_flush() -> None:
        nonlocal ret
        nonlocal incr_buf
        if incr_buf:
            ret += f"{'\t'*indent_lvl()}VALIDATE_NET_BYTES({incr_buf});\n"
            incr_buf = 0

    def gen_validate_size(path: idlutil.Path) -> None:
        nonlocal ret
        nonlocal incr_buf
        nonlocal indent_stack

        assert path.elems
        child = path.elems[-1]
        parent = path.elems[-2].typ if len(path.elems) > 1 else path.root
        assert isinstance(parent, idl.Struct)

        if child.in_versions < parent.in_versions:
            if line := cutil.ifdef_push(
                ifdef_lvl() + 1, c9util.ver_ifdef(child.in_versions)
            ):
                incr_flush()
                ret += line
                ret += (
                    f"{'\t'*indent_lvl()}if ({c9util.ver_cond(child.in_versions)}) {{\n"
                )
                indent_stack.append(IndentLevel(ifdef=True))
        if should_save_offset(parent, child):
            ret += f"{'\t'*indent_lvl()}uint32_t offsetof{''.join('_'+m.membname for m in path.elems)} = net_offset + {incr_buf};\n"
        if child.cnt:
            assert child.cnt.typ.static_size
            cnt_path = path.parent().add(child.cnt)
            incr_flush()
            if child.membname == "utf8":  # SPECIAL (string)
                # Yes, this is content-validation and "belongs" in
                # gen_validate_content(), not here.  But it's just
                # easier this way.
                ret += f"{'\t'*indent_lvl()}VALIDATE_NET_UTF8(LAST_U{child.cnt.typ.static_size*8}LE());\n"
                return
            if child.typ.static_size == 1:  # SPECIAL (zerocopy)
                ret += f"{'\t'*indent_lvl()}VALIDATE_NET_BYTES(LAST_U{child.cnt.typ.static_size*8}LE());\n"
                return
            loopdepth = sum(1 for elem in path.elems if elem.cnt)
            loopvar = chr(ord("i") + loopdepth - 1)
            ret += f"{'\t'*indent_lvl()}for ({c9util.typename(child.cnt.typ)} {loopvar} = 0, cnt = LAST_U{child.cnt.typ.static_size*8}LE(); {loopvar} < cnt; {loopvar}++) {{\n"
            indent_stack.append(IndentLevel(ifdef=False))
            ret += f"{'\t'*indent_lvl()}RESERVE_HOST_BYTES(sizeof({c9util.typename(child.typ)}));\n"
        if not isinstance(child.typ, idl.Struct):
            incr_buf += child.typ.static_size

    def gen_validate_content(path: idlutil.Path) -> None:
        nonlocal ret
        nonlocal incr_buf
        nonlocal indent_stack

        assert path.elems
        child = path.elems[-1]
        parent = path.elems[-2].typ if len(path.elems) > 1 else path.root
        assert isinstance(parent, idl.Struct)

        def lookup_sym(sym: str) -> str:
            if sym.startswith("&"):
                sym = sym[1:]
            return f"offsetof{''.join('_'+m.membname for m in path.elems[:-1])}_{sym}"

        if child.val:
            incr_flush()
            assert child.typ.static_size
            nbits = child.typ.static_size * 8
            nbits = child.typ.static_size * 8
            if nbits < 32 and any(
                isinstance(tok, idl.ExprSym)
                and (tok.symname == "end" or tok.symname.startswith("&"))
                for tok in child.val.tokens
            ):
                nbits = 32
            act = f"(uint{nbits}_t)GET_U{nbits}LE({lookup_sym(f'&{child.membname}')})"
            exp = f"(uint{nbits}_t)({c9util.idl_expr(child.val, lookup_sym)})"
            ret += f"{'\t'*indent_lvl()}if ({act} != {exp})\n"
            ret += f'{"\t"*(indent_lvl()+1)}return lib9p_errorf(ctx, LINUX_EBADMSG, "{path} value is wrong: actual: %"PRIu{nbits}" != correct:%"PRIu{nbits},\n'
            ret += f"{'\t'*(indent_lvl()+2)}{act}, {exp});\n"
        if child.max:
            incr_flush()
            assert child.typ.static_size
            nbits = child.typ.static_size * 8
            if nbits < 32 and any(
                isinstance(tok, idl.ExprSym)
                and (tok.symname == "end" or tok.symname.startswith("&"))
                for tok in child.max.tokens
            ):
                nbits = 32
            act = f"(uint{nbits}_t)GET_U{nbits}LE({lookup_sym(f'&{child.membname}')})"
            exp = f"(uint{nbits}_t)({c9util.idl_expr(child.max, lookup_sym)})"
            ret += f"{'\t'*indent_lvl()}if ({act} > {exp})\n"
            ret += f'{"\t"*(indent_lvl()+1)}return lib9p_errorf(ctx, LINUX_EBADMSG, "{path} value is too large: %"PRIu{nbits}" > %"PRIu{nbits},\n'
            ret += f"{'\t'*(indent_lvl()+2)}{act}, {exp});\n"
        if isinstance(child.typ, idl.Bitfield):
            incr_flush()
            nbytes = child.typ.static_size
            nbits = nbytes * 8
            act = f"GET_U{nbits}LE({lookup_sym(f'&{child.membname}')})"
            ret += f"{'\t'*indent_lvl()}if ({act} & ~{child.typ.typname}_masks[ctx->version])\n"
            ret += f'{"\t"*(indent_lvl()+1)}return lib9p_errorf(ctx, LINUX_EBADMSG, "unknown bits in {child.typ.typname} bitfield: %#0{nbytes*2}"PRIx{nbits},\n'
            ret += f"{'\t'*(indent_lvl()+2)}{act} & ~{child.typ.typname}_masks[ctx->version]);\n"

    def handle(
        path: idlutil.Path,
    ) -> tuple[idlutil.WalkCmd, typing.Callable[[], None]]:
        nonlocal ret
        nonlocal incr_buf
        nonlocal indent_stack
        indent_stack_len = len(indent_stack)
        pop_struct = path.elems[-1].typ if path.elems else path.root
        pop_path = path
        pop_indent_stack_len: int

        def pop() -> None:
            nonlocal ret
            nonlocal indent_stack
            nonlocal indent_stack_len
            nonlocal pop_struct
            nonlocal pop_path
            nonlocal pop_indent_stack_len
            if isinstance(pop_struct, idl.Struct):
                while len(indent_stack) > pop_indent_stack_len:
                    incr_flush()
                    ret += f"{'\t'*(indent_lvl()-1)}}}\n"
                    if indent_stack.pop().ifdef:
                        ret += cutil.ifdef_pop(ifdef_lvl())
                parent = pop_struct
                path = pop_path
                if should_save_end_offset(parent):
                    ret += f"{'\t'*indent_lvl()}uint32_t offsetof{''.join('_'+m.membname for m in path.elems)}_end = net_offset + {incr_buf};\n"
                for child in parent.members:
                    gen_validate_content(pop_path.add(child))
            while len(indent_stack) > indent_stack_len:
                if len(indent_stack) == indent_stack_len + 1 and indent_stack[-1].ifdef:
                    break
                incr_flush()
                ret += f"{'\t'*(indent_lvl()-1)}}}\n"
                if indent_stack.pop().ifdef:
                    ret += cutil.ifdef_pop(ifdef_lvl())

        if path.elems:
            gen_validate_size(path)

        pop_indent_stack_len = len(indent_stack)

        return idlutil.WalkCmd.KEEP_GOING, pop

    for typ in typs:
        if not (
            isinstance(typ, idl.Message) or typ.typname == "stat"
        ):  # SPECIAL (include stat)
            continue
        assert isinstance(typ, idl.Struct)
        ret += "\n"
        ret += cutil.ifdef_push(1, c9util.ver_ifdef(typ.in_versions))
        if typ.typname == "stat":  # SPECIAL (stat)
            ret += f"static ssize_t validate_{typ.typname}(struct lib9p_ctx *ctx, uint32_t net_size, uint8_t *net_bytes, uint32_t *ret_net_size) {{\n"
        else:
            ret += f"static ssize_t validate_{typ.typname}(struct lib9p_ctx *ctx, uint32_t net_size, uint8_t *net_bytes) {{\n"

        ret += "\tuint32_t net_offset = 0;\n"
        ret += f"\tssize_t host_size = sizeof({c9util.typename(typ)});\n"

        incr_buf = 0
        indent_stack = [IndentLevel(ifdef=True)]
        idlutil.walk(typ, handle)
        while len(indent_stack) > 1:
            incr_flush()
            ret += f"{'\t'*(indent_lvl()-1)}}}\n"
            if indent_stack.pop().ifdef:
                ret += cutil.ifdef_pop(ifdef_lvl())

        incr_flush()
        if typ.typname == "stat":  # SPECIAL (stat)
            ret += "\tif (ret_net_size)\n"
            ret += "\t\t*ret_net_size = net_offset;\n"
        ret += "\treturn (ssize_t)host_size;\n"
        ret += "}\n"
    ret += cutil.ifdef_pop(0)
    return ret