cmcfarlen commented on code in PR #13857:
URL: https://github.com/apache/trafficserver/pull/13857#discussion_r4354099853
##########
tools/hrw4u/src/procedures.py:
##########
@@ -20,10 +20,114 @@
from __future__ import annotations
+from dataclasses import dataclass, field
from pathlib import Path
+from typing import TYPE_CHECKING
+
+from hrw4u.errors import Hrw4uSyntaxError, SourceMap
+
+if TYPE_CHECKING:
+ import hrw4u.ast_nodes as ast
def resolve_use_path(spec: str, search_paths: list[Path]) -> Path | None:
"""Resolve 'foo::bar::Baz' to <search_path>/foo/bar/Baz.hrw4u."""
rel = Path(*spec.split('::')).with_suffix('.hrw4u')
return next((base / rel for base in search_paths if (base /
rel).is_file()), None)
+
+
+@dataclass(frozen=True, slots=True)
+class ProcEntry:
+ name: str
+ params: tuple[ast.ProcParam, ...]
+ body: tuple[ast.BodyNode, ...]
+ source_file: str
+
+
+@dataclass
+class ProcRegistry:
+ """Procedure declarations, keyed by qualified name."""
+ source_map: SourceMap
+ _sigs: dict[str, ProcEntry] = field(default_factory=dict, init=False)
+ _loaded: set[str] = field(default_factory=set, init=False)
+
+ def get(self, name: str) -> ProcEntry | None:
+ return self._sigs.get(name)
+
+ def register(self, decl: ast.ProcedureDecl, expected_ns: str | None =
None, use_spec: str | None = None) -> None:
+ span = decl.span
+ if '::' not in decl.name:
+ raise self.source_map.error_at(span, f"procedure name
'{decl.name}' must be qualified (e.g. 'ns::name')")
+ if expected_ns and not decl.name.startswith(expected_ns):
+ namespace = expected_ns.removesuffix('::')
+ raise self.source_map.error_at(
+ span, f"procedure '{decl.name}' does not match namespace
'{namespace}' (expected from 'use {use_spec}')")
+ if existing := self._sigs.get(decl.name):
+ raise self.source_map.error_at(span, f"procedure '{decl.name}'
already declared in {existing.source_file}")
+
+ seen_default = False
+ for param in decl.params:
+ if param.default is None and seen_default:
Review Comment:
The CST emitter only runs this check in `visitProcedureDecl`, for inline
declarations. `_load_proc_file` never ran it for procedures loaded through
`use`. So a library file with
```
procedure syn::p($a = "x", $b) {
inbound.req.X-A = "$a$b";
}
```
called as `syn::p("1", "2")` compiles today. Sema rejects it with this
error, then reports `unknown procedure` at the call. After the switchover, any
config that `use`s a library like that stops compiling. Same request as the
other comment: add a parity case and either match the CST behaviour or record
the divergence.
##########
tools/hrw4u/src/sema.py:
##########
@@ -0,0 +1,662 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import annotations
+
+import re
+from contextlib import contextmanager
+from dataclasses import replace
+from pathlib import Path
+from typing import Callable
+
+import hrw4u.ast_nodes as ast
+import hrw4u.sema_nodes as ir
+import hrw4u.types as types
+from hrw4u.common import RegexPatterns, SystemDefaults, parse_mods,
split_interpolation_mods
+from hrw4u.debugging import Dbg
+from hrw4u.errors import ErrorCollector, Hrw4uSyntaxError, SourceMap,
SymbolResolutionError, Warning, hrw4u_error_at
+from hrw4u.procedures import ProcEntry, ProcRegistry
+from hrw4u.sandbox import SandboxConfig, SandboxDenialError
+from hrw4u.states import CondState, SectionType, VALUE_MODIFIERS
+from hrw4u.symbols import SymbolResolver
+from hrw4u.validation import Validator
+
+_VAR_SECTIONS = {
+ "txn": ("Variable", "VARS", types.VarScope.TXN),
+ "session": ("Session variable", "SESSION_VARS", types.VarScope.SESSION),
+}
+
+_PARAM_REF = re.compile(r'\$([a-zA-Z_][a-zA-Z0-9_-]*)')
+
+
+def _strip_quotes(text: str) -> str:
+ if text.startswith('"') and text.endswith('"'):
+ return text[1:-1]
+ return text
+
+
+# _split_args and _unescape_double_braces are copied from HRW4UVisitor rather
than shared: its copies go with the CST emitter.
+def _split_args(arg_str: str) -> list[str]:
+ if not arg_str.strip():
+ return []
+
+ args = []
+ current_arg = []
+ paren_depth = 0
+ in_quotes = False
+ quote_char = None
+
+ for i, char in enumerate(arg_str):
+ if not in_quotes:
+ if char in ('"', "'"):
+ in_quotes = True
+ quote_char = char
+ current_arg.append(char)
+ elif char == '(':
+ paren_depth += 1
+ current_arg.append(char)
+ elif char == ')':
+ paren_depth -= 1
+ current_arg.append(char)
+ elif char == ',' and paren_depth == 0:
+ args.append(''.join(current_arg).strip())
+ current_arg = []
+ else:
+ current_arg.append(char)
+ else:
+ current_arg.append(char)
+ if char == quote_char and (i == 0 or arg_str[i - 1] != '\\'):
+ in_quotes = False
+ quote_char = None
+
+ if current_arg:
+ args.append(''.join(current_arg).strip())
+
+ return args
+
+
+def _unescape_double_braces(escaped: str) -> str:
+ """`{{...}}` in a string literal: the payload as literal `{...}`,
unescaped quotes escaped."""
+ out: list[str] = []
+ backslashes = 0
+ for char in escaped[2:-2]:
+ if char == '"' and backslashes % 2 == 0:
+ out.append('\\')
+ out.append(char)
+ if char == '\\':
+ backslashes += 1
+ else:
+ backslashes = 0
+ return "{" + "".join(out) + "}"
+
+
+class Sema:
+ """Resolves an HRW4U AST into a backend-neutral tree.
+
+ Sandbox checks run during resolution, here and in SymbolResolver.lookup_*,
rather than as one
+ pass over the finished tree: the CST emitter checks as it goes and drops
what it denies, and
+ parity holds sema to the same diagnostics in the same order. Once the CLI
routes through sema
+ and the CST emitter is gone, they can move to a single pass after
resolution.
+ """
+
+ def __init__(
+ self,
+ *,
+ debug: bool = SystemDefaults.DEFAULT_DEBUG,
+ error_collector: ErrorCollector | None = None,
+ preserve_comments: bool = True,
+ proc_search_paths: list[Path] | None = None,
+ sandbox: SandboxConfig | None = None) -> None:
+ self.error_collector = error_collector
+ self.preserve_comments = preserve_comments
+ self.source_map = SourceMap()
+ self._sandbox = sandbox or SandboxConfig.empty()
+ self._dbg = Dbg(debug)
+ self.symbol_resolver = SymbolResolver(debug, sandbox=self._sandbox,
dbg=self._dbg)
+ self._proc_search_paths: list[Path] = list(proc_search_paths or [])
+ self._section: SectionType | None = None
+ self.procs = ProcRegistry(source_map=self.source_map)
+ self._proc_stack: list[str] = []
+
+ def _report(self, span: ast.Span, exc: Exception, *, note: str | None =
None) -> None:
+ error = hrw4u_error_at(span, exc, self.source_map.line_at(span))
+ if note:
+ error.add_note(note)
+ if self.error_collector is None:
+ raise error
+ self.error_collector.add_error(error)
+ if isinstance(exc, SandboxDenialError) and exc.sandbox_message:
+ self.error_collector.set_sandbox_message(exc.sandbox_message)
+
+ @contextmanager
+ def trap(self, span: ast.Span, *, note: str | None = None):
+ try:
+ yield
+ except Exception as exc:
+ self._report(span, exc, note=note)
+
+ def _warn(self, span: ast.Span, message: str) -> None:
+ if self.error_collector is not None:
+ self.error_collector.add_warning(Warning.from_span(span, message,
self.source_map.line_at(span)))
+ if self._sandbox.message:
+ self.error_collector.set_sandbox_message(self._sandbox.message)
+
+ def _drain_resolver_warnings(self, span: ast.Span) -> None:
+ for warning in self.symbol_resolver.drain_warnings():
+ self._warn(span, warning)
+
+ def _unbound_param(self, name: str, span: ast.Span) -> Hrw4uSyntaxError:
+ return self.source_map.error_at(span, f"'${name}' used outside
procedure context")
+
+ def _sandbox_check(self, span: ast.Span, check_fn: Callable[[], str |
None]) -> bool:
+ try:
+ if warning := check_fn():
+ self._warn(span, warning)
+ return True
+ except SandboxDenialError as exc:
+ self._report(span, exc)
+ return False
+
+ def analyze(self, tree: ast.HRW4UAST) -> ir.Program:
+ items: list[ir.ResolvedSection | ir.VarsBlock | ir.CommentStmt] = []
+ seen_sections = False
+
+ for node in tree.body:
+ match node:
+ case ast.Comment():
+ if self.preserve_comments:
+ items.append(ir.CommentStmt(text=node.text,
span=node.span))
+ case ast.VarSection():
+ seen_sections = True
+ items.append(self._declare_vars(node))
+ case ast.UseDirective():
+ if seen_sections:
+ self._report(node.span, ValueError("'use' directives
must appear before any section blocks"))
+ continue
+ with self.trap(node.span):
+ self.procs.load(node.spec, node.span,
self._proc_search_paths)
+ case ast.ProcedureDecl():
+ if seen_sections:
+ self._report(node.span, ValueError("'procedure'
declarations must appear before any section blocks"))
+ continue
+ with self.trap(node.span):
+ self.procs.register(node)
+ case ast.Section():
+ seen_sections = True
+ if section := self._section_of(node):
+ items.append(section)
+
+ return ir.Program(items=tuple(items))
+
+ def _section_of(self, node: ast.Section) -> ir.ResolvedSection | None:
+ with self.trap(node.span):
+ if warning := self._sandbox.check_section(node.type):
+ self._warn(node.span, warning)
+ try:
+ self._section = SectionType(node.type)
+ except ValueError:
+ valid = ', '.join(s.value for s in SectionType)
+ raise ValueError(f"Invalid section name: '{node.type}'. Valid
sections: {valid}") from None
+
+ hook = self._section.hook_name
+ self._dbg(f"`{node.type}' -> `{hook}'")
+ return ir.ResolvedSection(section=self._section,
body=self._body(node.body), span=node.span)
+ return None
+
+ def _body(self, nodes: tuple[ast.BodyNode, ...]) -> tuple[ir.Stmt, ...]:
+ out: list[ir.Stmt] = []
+ for node in nodes:
+ # Per statement, not per section: a denial or a resolution failure
costs that
+ # statement and not everything around it.
+ with self.trap(node.span):
+ self._body_item(node, out)
+ return tuple(out)
+
+ def _body_item(self, node: ast.BodyNode, out: list[ir.Stmt]) -> None:
+ match node:
+ case ast.Comment():
+ if self.preserve_comments:
+ out.append(ir.CommentStmt(text=node.text, span=node.span))
+ case ast.Break():
+ if warning := self._sandbox.check_language("break"):
+ self._warn(node.span, warning)
+ out.append(ir.BreakStmt(span=node.span))
+ case ast.Assignment():
+ if stmt := self._assignment(node):
+ out.append(stmt)
+ case ast.FunctionCall() if self.procs.get(node.name) is not None:
+ with self._inlining(node) as inlined:
+ if inlined is None:
+ return
+ # The first source item is resolved alone so that Inlined
can say whether body[0] is it.
+ head, rest = self._body(inlined[:1]),
self._body(inlined[1:])
+ opens_with_statement = bool(head) and not
isinstance(head[0], (ir.CommentStmt, ir.Inlined))
+ out.append(ir.Inlined(body=head + rest,
opens_with_statement=opens_with_statement, span=node.span))
+ case ast.FunctionCall():
+ if stmt := self._statement_call(node):
+ out.append(stmt)
+ case ast.IfBlock():
+ out.append(self._if_block(node))
+
+ def _if_block(self, node: ast.IfBlock) -> ir.If:
+ # Source order, so that without a collector the error raised is the
first one in the file.
+ cond = self._condition(node.condition)
+ body = self._body(node.body)
+
+ elifs = []
+ for arm in node.elif_branches:
+ if self._sandbox_check(arm.span, lambda:
self._sandbox.check_language("elif")):
+ elifs.append(ir.ElifArm(cond=self._condition(arm.condition),
body=self._body(arm.body), span=arm.span))
+
+ else_body: tuple[ir.Stmt, ...] = ()
+ else_span = node.else_span or node.span
+ has_else = node.has_else and self._sandbox_check(else_span, lambda:
self._sandbox.check_language("else"))
+ if has_else:
+ else_body = self._body(node.else_body)
+
+ return ir.If(cond=cond, body=body, elifs=tuple(elifs),
else_body=else_body, span=node.span, has_else=has_else)
+
+ def _condition(self, node: ast.ConditionExpr) -> ir.Cond:
+ match node:
+ case ast.LogicalOp():
+ if node.operator == "||":
+ modifier, combine = "OR", ir.Or
+ else:
+ modifier, combine = "AND", ir.And
+ if not self._sandbox_check(node.span, lambda:
self._sandbox.check_modifier(modifier)):
+ return ir.Unresolved(span=node.span)
+ left, right = self._condition(node.left),
self._condition(node.right)
+ return combine(left=left, right=right, span=node.span)
+ case ast.NotOp():
+ if not self._sandbox_check(node.span, lambda:
self._sandbox.check_modifier("NOT")):
+ return ir.Unresolved(span=node.span)
+ return ir.Not(inner=self._condition(node.operand),
span=node.span)
+ case ast.Group():
+ return ir.Group(inner=self._condition(node.inner),
span=node.span)
+ case ast.BoolLiteral():
+ return ir.BoolCond(value=node.value, span=node.span)
+ case ast.FunctionCall():
+ return self._func_cond(node)
+ case ast.IdentCondition():
+ return self._truth(node)
+ case ast.Comparison():
+ return self._test(node)
+ raise SymbolResolutionError("condition", f"Unsupported condition
{node!r}")
+
+ def _func_cond(self, node: ast.FunctionCall) -> ir.Cond:
+ with self.trap(node.span):
+ # Source text, not resolved values: a `{...}` in a condition
function's string argument
+ # is not interpolated.
+ args = [self._source_text(a, node.span) for a in node.args]
+ # Validate before stripping quotes: quoted_or_simple() needs to
see the quotes on values
+ # such as file paths.
+ key, entry = self.symbol_resolver.lookup_function(node.name, args)
+ self._drain_resolver_warnings(node.span)
+ return ir.FuncCond(key=key, entry=entry,
args=tuple(ir.RawArg(text=_strip_quotes(a)) for a in args), span=node.span)
+ return ir.Unresolved(span=node.span)
+
+ def _truth(self, node: ast.IdentCondition) -> ir.Cond:
+ with self.trap(node.span):
+ ref, existence = self._reference(node.name, node.span)
+ return ir.Truth(ref=ref, existence=existence, span=node.span)
+ return ir.Unresolved(span=node.span)
+
+ def _test(self, node: ast.Comparison) -> ir.Cond:
+ with self.trap(node.span):
+ if isinstance(node.left, ast.FunctionCall):
+ func = self._func_cond(node.left)
+ if not isinstance(func, ir.FuncCond):
+ return ir.Unresolved(span=node.span) # _func_cond already
reported
+ ref = ir.BuiltinRef.from_function(func.key, func.entry,
[a.text for a in func.args])
+ else:
+ ref, _ = self._reference(node.left.raw, node.span)
+
+ op = ir.CmpOp(node.operator)
+
+ if op in ir.NEGATED_OPS and not self._sandbox_check(node.span,
lambda: self._sandbox.check_modifier("NOT")):
+ return ir.Unresolved(span=node.span)
+
+ if op in (ir.CmpOp.IN, ir.CmpOp.NOT_IN):
+ rhs = self._operand(node.right, node.span)
+ if not self._sandbox_check(node.span, lambda:
self._sandbox.check_language("in")):
+ return ir.Unresolved(span=node.span)
+ elif op in (ir.CmpOp.MATCH, ir.CmpOp.NOT_MATCH):
+ # Trapped here rather than by the enclosing trap: a bad
pattern must not cost the
+ # modifier diagnostics below.
+ with self.trap(node.right.span or node.span):
+ Validator.regex_pattern()(node.right.raw)
+ rhs = ir.RegexLit(text=node.right.raw)
+ else:
+ rhs = self._operand(node.right, node.span)
+
+ # A denied, unknown or conflicting modifier drops itself and every
modifier after it,
+ # but the comparison survives. Validated here, not left to the
emitter, so that it is
+ # a collected error rather than a raise out of the compile.
+ mods: list[str] = []
+ mod_span = node.modifier_span or node.span
+ state = CondState()
+ for mod in node.modifiers:
+ if not self._sandbox_check(mod_span, lambda m=mod:
self._sandbox.check_modifier(m.upper())):
+ break
+ try:
+ state.add_modifier(mod)
+ except SymbolResolutionError as exc:
+ self._report(mod_span, exc)
+ break
+ mods.append(mod.upper())
+
+ return ir.Test(ref=ref, op=op, rhs=rhs, modifiers=tuple(mods),
span=node.span)
+ return ir.Unresolved(span=node.span)
+
+ def _reference(self, name: str, span: ast.Span) -> tuple[ir.Ref, bool]:
+ if not name:
+ raise SymbolResolutionError("identifier", "Missing or empty
identifier text")
+
+ try:
+ found = self.symbol_resolver.lookup_condition(name, self._section)
+ except SandboxDenialError:
+ raise
+ except SymbolResolutionError:
+ # Only an unqualified name can be a variable; a dotted or
qualified one keeps the resolver's error.
+ if '.' not in name and ':' not in name:
+ error = SymbolResolutionError(
+ "identifier", f"Undefined variable: '{name}'. Variables
must be declared in a VARS section.")
+ if suggestions :=
self.symbol_resolver.get_variable_suggestions(name, self._section):
+ error.add_symbol_suggestion(suggestions)
+ raise error from None
+ raise
+
+ self._drain_resolver_warnings(span)
+
+ if found.symbol is not None:
+ return ir.VarRef(symbol=found.symbol), False
+ return ir.BuiltinRef(key=found.key, entry=found.entry, arg=found.arg),
found.existence
+
+ def _assignment(self, node: ast.Assignment) -> ir.Stmt | None:
+ with self.trap(node.span):
+ value = self._operand(node.value, node.span)
+ value_text = ir.value_text(value)
+ add = node.operator == "+="
+ found = self.symbol_resolver.lookup_operator(node.name,
value_text, self._section, add=add)
+ self._drain_resolver_warnings(node.span)
+
+ if found.symbol is not None:
+ if isinstance(node.value, ast.IdentValue) and (rhs_symbol :=
self.symbol_resolver.symbol_for(node.value.raw)):
+ value = ir.VarRef(symbol=rhs_symbol)
+ return ir.SetVar(symbol=found.symbol, value=value,
span=node.span)
+
+ if add:
+ variant = ir.OpVariant.ADD
+ elif value_text == '""':
+ variant = ir.OpVariant.REMOVE
+ else:
+ variant = ir.OpVariant.SET
+
+ return ir.SetOp(
+ key=found.key, entry=found.entry, qualifier=found.qualifier,
value=value, variant=variant, span=node.span)
+ return None
+
+ def _statement_call(self, node: ast.FunctionCall) -> ir.Stmt | None:
+ with self.trap(node.span):
+ if '::' in node.name:
+ raise self.source_map.error_at(node.span, f"unknown procedure
'{node.name}': not loaded via 'use'")
+ args = [self._value(a, node.span) for a in node.args]
+ arg_texts = [ir.value_text(a) for a in args]
+ key, entry =
self.symbol_resolver.lookup_statement_function(node.name, arg_texts,
self._section)
+ self._drain_resolver_warnings(node.span)
+ return ir.CallOp(key=key, entry=entry,
args=tuple(map(self._normalized_arg, args, arg_texts)), span=node.span)
+ return None
+
+ @staticmethod
+ def _normalized_arg(arg: ir.RValue, text: str) -> ir.RValue:
+ """Validators normalize argument text in place; an argument they left
alone keeps its resolved form."""
+ if text == ir.value_text(arg):
+ return arg
+ return ir.RawArg(text=text)
+
+ @contextmanager
+ def _inlining(self, call: ast.FunctionCall):
+ sig = self.procs.get(call.name)
+ if call.name in self._proc_stack:
+ self._report(call.span, ValueError(f"circular procedure call: {'
-> '.join([*self._proc_stack, call.name])}"))
+ yield None
+ return
+
+ bindings = self._bind(sig, call)
+ if bindings is None:
+ yield None
+ return
+
+ body = tuple(self._substitute(node, bindings) for node in sig.body)
+ self._proc_stack.append(call.name)
+ try:
+ yield body
+ finally:
+ self._proc_stack.pop()
+
+ def _bind(self, sig: ProcEntry, call: ast.FunctionCall) -> dict[str, str]
| None:
+ with self.trap(call.span):
+ args = [self._binding_text(a, call.span) for a in call.args]
+ required = sum(1 for p in sig.params if p.default is None)
+ if not (required <= len(args) <= len(sig.params)):
+ expected = str(len(sig.params))
+ if required < len(sig.params):
+ expected = f"{required}-{len(sig.params)}"
+ raise ValueError(f"procedure '{sig.name}': expected {expected}
arg(s), got {len(args)}")
+
+ bindings = dict(zip((p.name for p in sig.params), args))
+ for param in sig.params[len(args):]:
+ bindings[param.name] = self._default_text(param.default,
call.span)
+ return bindings
+ return None
+
+ def _default_text(self, value: ast.ValueExpr, span: ast.Span) -> str:
+ """A string default loses its quotes but, unlike a call argument, is
not interpolated."""
+ if isinstance(value, ast.LiteralStringValue):
+ return value.raw
+ return self._binding_text(value, span)
+
+ def _binding_text(self, value, span: ast.Span) -> str:
+ """A binding is plain text: a string literal contributes its inner
text."""
+ if isinstance(value, ast.LiteralStringValue):
+ return ir.value_text(self._template(value.raw, span))[1:-1]
+ if isinstance(value, ast.ParamRef):
+ # _substitute replaced every bound `$param`, so one left here is
unbound.
+ raise self._unbound_param(value.raw, value.span or span)
Review Comment:
A parameter default that refers to another parameter diverges from the CST
emitter, and the parity corpus doesn't cover it:
```
procedure syn::inner($x = $a) {
inbound.req.X-A = $x;
}
procedure syn::outer($a) {
syn::inner();
}
```
called as `syn::outer("v");`. The CST emitter evaluates the default while
`outer`'s bindings are still active, so `$x` takes the caller's `$a` and the
file compiles. Sema reports `'$a' used outside procedure context` here.
Rejecting it may well be the better rule. Either way, please add it as a
synthetic parity case, and if Sema's behaviour is the intended one, list it
with `COLLECTS_INSTEAD_OF_RAISING` or something similar, so the switchover
doesn't change it silently.
##########
tools/hrw4u/src/sema.py:
##########
@@ -0,0 +1,662 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import annotations
+
+import re
+from contextlib import contextmanager
+from dataclasses import replace
+from pathlib import Path
+from typing import Callable
+
+import hrw4u.ast_nodes as ast
+import hrw4u.sema_nodes as ir
+import hrw4u.types as types
+from hrw4u.common import RegexPatterns, SystemDefaults, parse_mods,
split_interpolation_mods
+from hrw4u.debugging import Dbg
+from hrw4u.errors import ErrorCollector, Hrw4uSyntaxError, SourceMap,
SymbolResolutionError, Warning, hrw4u_error_at
+from hrw4u.procedures import ProcEntry, ProcRegistry
+from hrw4u.sandbox import SandboxConfig, SandboxDenialError
+from hrw4u.states import CondState, SectionType, VALUE_MODIFIERS
+from hrw4u.symbols import SymbolResolver
+from hrw4u.validation import Validator
+
+_VAR_SECTIONS = {
+ "txn": ("Variable", "VARS", types.VarScope.TXN),
+ "session": ("Session variable", "SESSION_VARS", types.VarScope.SESSION),
+}
+
+_PARAM_REF = re.compile(r'\$([a-zA-Z_][a-zA-Z0-9_-]*)')
+
+
+def _strip_quotes(text: str) -> str:
+ if text.startswith('"') and text.endswith('"'):
+ return text[1:-1]
+ return text
+
+
+# _split_args and _unescape_double_braces are copied from HRW4UVisitor rather
than shared: its copies go with the CST emitter.
+def _split_args(arg_str: str) -> list[str]:
+ if not arg_str.strip():
+ return []
+
+ args = []
+ current_arg = []
+ paren_depth = 0
+ in_quotes = False
+ quote_char = None
+
+ for i, char in enumerate(arg_str):
+ if not in_quotes:
+ if char in ('"', "'"):
+ in_quotes = True
+ quote_char = char
+ current_arg.append(char)
+ elif char == '(':
+ paren_depth += 1
+ current_arg.append(char)
+ elif char == ')':
+ paren_depth -= 1
+ current_arg.append(char)
+ elif char == ',' and paren_depth == 0:
+ args.append(''.join(current_arg).strip())
+ current_arg = []
+ else:
+ current_arg.append(char)
+ else:
+ current_arg.append(char)
+ if char == quote_char and (i == 0 or arg_str[i - 1] != '\\'):
+ in_quotes = False
+ quote_char = None
+
+ if current_arg:
+ args.append(''.join(current_arg).strip())
+
+ return args
+
+
+def _unescape_double_braces(escaped: str) -> str:
+ """`{{...}}` in a string literal: the payload as literal `{...}`,
unescaped quotes escaped."""
+ out: list[str] = []
+ backslashes = 0
+ for char in escaped[2:-2]:
+ if char == '"' and backslashes % 2 == 0:
+ out.append('\\')
+ out.append(char)
+ if char == '\\':
+ backslashes += 1
+ else:
+ backslashes = 0
+ return "{" + "".join(out) + "}"
+
+
+class Sema:
+ """Resolves an HRW4U AST into a backend-neutral tree.
+
+ Sandbox checks run during resolution, here and in SymbolResolver.lookup_*,
rather than as one
+ pass over the finished tree: the CST emitter checks as it goes and drops
what it denies, and
+ parity holds sema to the same diagnostics in the same order. Once the CLI
routes through sema
+ and the CST emitter is gone, they can move to a single pass after
resolution.
+ """
+
+ def __init__(
+ self,
+ *,
+ debug: bool = SystemDefaults.DEFAULT_DEBUG,
+ error_collector: ErrorCollector | None = None,
+ preserve_comments: bool = True,
+ proc_search_paths: list[Path] | None = None,
+ sandbox: SandboxConfig | None = None) -> None:
+ self.error_collector = error_collector
+ self.preserve_comments = preserve_comments
+ self.source_map = SourceMap()
+ self._sandbox = sandbox or SandboxConfig.empty()
+ self._dbg = Dbg(debug)
+ self.symbol_resolver = SymbolResolver(debug, sandbox=self._sandbox,
dbg=self._dbg)
+ self._proc_search_paths: list[Path] = list(proc_search_paths or [])
+ self._section: SectionType | None = None
+ self.procs = ProcRegistry(source_map=self.source_map)
+ self._proc_stack: list[str] = []
+
+ def _report(self, span: ast.Span, exc: Exception, *, note: str | None =
None) -> None:
+ error = hrw4u_error_at(span, exc, self.source_map.line_at(span))
+ if note:
+ error.add_note(note)
+ if self.error_collector is None:
+ raise error
+ self.error_collector.add_error(error)
+ if isinstance(exc, SandboxDenialError) and exc.sandbox_message:
+ self.error_collector.set_sandbox_message(exc.sandbox_message)
+
+ @contextmanager
+ def trap(self, span: ast.Span, *, note: str | None = None):
+ try:
+ yield
+ except Exception as exc:
+ self._report(span, exc, note=note)
+
+ def _warn(self, span: ast.Span, message: str) -> None:
+ if self.error_collector is not None:
+ self.error_collector.add_warning(Warning.from_span(span, message,
self.source_map.line_at(span)))
+ if self._sandbox.message:
+ self.error_collector.set_sandbox_message(self._sandbox.message)
+
+ def _drain_resolver_warnings(self, span: ast.Span) -> None:
+ for warning in self.symbol_resolver.drain_warnings():
+ self._warn(span, warning)
+
+ def _unbound_param(self, name: str, span: ast.Span) -> Hrw4uSyntaxError:
+ return self.source_map.error_at(span, f"'${name}' used outside
procedure context")
+
+ def _sandbox_check(self, span: ast.Span, check_fn: Callable[[], str |
None]) -> bool:
+ try:
+ if warning := check_fn():
+ self._warn(span, warning)
+ return True
+ except SandboxDenialError as exc:
+ self._report(span, exc)
+ return False
+
+ def analyze(self, tree: ast.HRW4UAST) -> ir.Program:
+ items: list[ir.ResolvedSection | ir.VarsBlock | ir.CommentStmt] = []
+ seen_sections = False
+
+ for node in tree.body:
+ match node:
+ case ast.Comment():
+ if self.preserve_comments:
+ items.append(ir.CommentStmt(text=node.text,
span=node.span))
+ case ast.VarSection():
+ seen_sections = True
+ items.append(self._declare_vars(node))
+ case ast.UseDirective():
+ if seen_sections:
+ self._report(node.span, ValueError("'use' directives
must appear before any section blocks"))
+ continue
+ with self.trap(node.span):
+ self.procs.load(node.spec, node.span,
self._proc_search_paths)
+ case ast.ProcedureDecl():
+ if seen_sections:
+ self._report(node.span, ValueError("'procedure'
declarations must appear before any section blocks"))
+ continue
+ with self.trap(node.span):
+ self.procs.register(node)
+ case ast.Section():
+ seen_sections = True
+ if section := self._section_of(node):
+ items.append(section)
+
+ return ir.Program(items=tuple(items))
+
+ def _section_of(self, node: ast.Section) -> ir.ResolvedSection | None:
+ with self.trap(node.span):
+ if warning := self._sandbox.check_section(node.type):
+ self._warn(node.span, warning)
+ try:
+ self._section = SectionType(node.type)
+ except ValueError:
+ valid = ', '.join(s.value for s in SectionType)
+ raise ValueError(f"Invalid section name: '{node.type}'. Valid
sections: {valid}") from None
+
+ hook = self._section.hook_name
+ self._dbg(f"`{node.type}' -> `{hook}'")
+ return ir.ResolvedSection(section=self._section,
body=self._body(node.body), span=node.span)
+ return None
+
+ def _body(self, nodes: tuple[ast.BodyNode, ...]) -> tuple[ir.Stmt, ...]:
+ out: list[ir.Stmt] = []
+ for node in nodes:
+ # Per statement, not per section: a denial or a resolution failure
costs that
+ # statement and not everything around it.
+ with self.trap(node.span):
+ self._body_item(node, out)
+ return tuple(out)
+
+ def _body_item(self, node: ast.BodyNode, out: list[ir.Stmt]) -> None:
+ match node:
+ case ast.Comment():
+ if self.preserve_comments:
+ out.append(ir.CommentStmt(text=node.text, span=node.span))
+ case ast.Break():
+ if warning := self._sandbox.check_language("break"):
+ self._warn(node.span, warning)
+ out.append(ir.BreakStmt(span=node.span))
+ case ast.Assignment():
+ if stmt := self._assignment(node):
+ out.append(stmt)
+ case ast.FunctionCall() if self.procs.get(node.name) is not None:
+ with self._inlining(node) as inlined:
+ if inlined is None:
+ return
+ # The first source item is resolved alone so that Inlined
can say whether body[0] is it.
+ head, rest = self._body(inlined[:1]),
self._body(inlined[1:])
+ opens_with_statement = bool(head) and not
isinstance(head[0], (ir.CommentStmt, ir.Inlined))
+ out.append(ir.Inlined(body=head + rest,
opens_with_statement=opens_with_statement, span=node.span))
+ case ast.FunctionCall():
+ if stmt := self._statement_call(node):
+ out.append(stmt)
+ case ast.IfBlock():
+ out.append(self._if_block(node))
+
+ def _if_block(self, node: ast.IfBlock) -> ir.If:
+ # Source order, so that without a collector the error raised is the
first one in the file.
+ cond = self._condition(node.condition)
+ body = self._body(node.body)
+
+ elifs = []
+ for arm in node.elif_branches:
+ if self._sandbox_check(arm.span, lambda:
self._sandbox.check_language("elif")):
+ elifs.append(ir.ElifArm(cond=self._condition(arm.condition),
body=self._body(arm.body), span=arm.span))
+
+ else_body: tuple[ir.Stmt, ...] = ()
+ else_span = node.else_span or node.span
+ has_else = node.has_else and self._sandbox_check(else_span, lambda:
self._sandbox.check_language("else"))
+ if has_else:
+ else_body = self._body(node.else_body)
+
+ return ir.If(cond=cond, body=body, elifs=tuple(elifs),
else_body=else_body, span=node.span, has_else=has_else)
+
+ def _condition(self, node: ast.ConditionExpr) -> ir.Cond:
+ match node:
+ case ast.LogicalOp():
+ if node.operator == "||":
+ modifier, combine = "OR", ir.Or
+ else:
+ modifier, combine = "AND", ir.And
+ if not self._sandbox_check(node.span, lambda:
self._sandbox.check_modifier(modifier)):
+ return ir.Unresolved(span=node.span)
+ left, right = self._condition(node.left),
self._condition(node.right)
+ return combine(left=left, right=right, span=node.span)
+ case ast.NotOp():
+ if not self._sandbox_check(node.span, lambda:
self._sandbox.check_modifier("NOT")):
+ return ir.Unresolved(span=node.span)
+ return ir.Not(inner=self._condition(node.operand),
span=node.span)
+ case ast.Group():
+ return ir.Group(inner=self._condition(node.inner),
span=node.span)
+ case ast.BoolLiteral():
+ return ir.BoolCond(value=node.value, span=node.span)
+ case ast.FunctionCall():
+ return self._func_cond(node)
+ case ast.IdentCondition():
+ return self._truth(node)
+ case ast.Comparison():
+ return self._test(node)
+ raise SymbolResolutionError("condition", f"Unsupported condition
{node!r}")
+
+ def _func_cond(self, node: ast.FunctionCall) -> ir.Cond:
+ with self.trap(node.span):
+ # Source text, not resolved values: a `{...}` in a condition
function's string argument
+ # is not interpolated.
+ args = [self._source_text(a, node.span) for a in node.args]
+ # Validate before stripping quotes: quoted_or_simple() needs to
see the quotes on values
+ # such as file paths.
+ key, entry = self.symbol_resolver.lookup_function(node.name, args)
+ self._drain_resolver_warnings(node.span)
+ return ir.FuncCond(key=key, entry=entry,
args=tuple(ir.RawArg(text=_strip_quotes(a)) for a in args), span=node.span)
+ return ir.Unresolved(span=node.span)
+
+ def _truth(self, node: ast.IdentCondition) -> ir.Cond:
+ with self.trap(node.span):
+ ref, existence = self._reference(node.name, node.span)
+ return ir.Truth(ref=ref, existence=existence, span=node.span)
+ return ir.Unresolved(span=node.span)
+
+ def _test(self, node: ast.Comparison) -> ir.Cond:
+ with self.trap(node.span):
+ if isinstance(node.left, ast.FunctionCall):
+ func = self._func_cond(node.left)
+ if not isinstance(func, ir.FuncCond):
+ return ir.Unresolved(span=node.span) # _func_cond already
reported
+ ref = ir.BuiltinRef.from_function(func.key, func.entry,
[a.text for a in func.args])
+ else:
+ ref, _ = self._reference(node.left.raw, node.span)
+
+ op = ir.CmpOp(node.operator)
+
+ if op in ir.NEGATED_OPS and not self._sandbox_check(node.span,
lambda: self._sandbox.check_modifier("NOT")):
+ return ir.Unresolved(span=node.span)
+
+ if op in (ir.CmpOp.IN, ir.CmpOp.NOT_IN):
+ rhs = self._operand(node.right, node.span)
+ if not self._sandbox_check(node.span, lambda:
self._sandbox.check_language("in")):
+ return ir.Unresolved(span=node.span)
+ elif op in (ir.CmpOp.MATCH, ir.CmpOp.NOT_MATCH):
+ # Trapped here rather than by the enclosing trap: a bad
pattern must not cost the
+ # modifier diagnostics below.
+ with self.trap(node.right.span or node.span):
+ Validator.regex_pattern()(node.right.raw)
+ rhs = ir.RegexLit(text=node.right.raw)
+ else:
+ rhs = self._operand(node.right, node.span)
+
+ # A denied, unknown or conflicting modifier drops itself and every
modifier after it,
+ # but the comparison survives. Validated here, not left to the
emitter, so that it is
+ # a collected error rather than a raise out of the compile.
+ mods: list[str] = []
+ mod_span = node.modifier_span or node.span
+ state = CondState()
+ for mod in node.modifiers:
+ if not self._sandbox_check(mod_span, lambda m=mod:
self._sandbox.check_modifier(m.upper())):
+ break
+ try:
+ state.add_modifier(mod)
+ except SymbolResolutionError as exc:
+ self._report(mod_span, exc)
+ break
+ mods.append(mod.upper())
+
+ return ir.Test(ref=ref, op=op, rhs=rhs, modifiers=tuple(mods),
span=node.span)
+ return ir.Unresolved(span=node.span)
+
+ def _reference(self, name: str, span: ast.Span) -> tuple[ir.Ref, bool]:
+ if not name:
+ raise SymbolResolutionError("identifier", "Missing or empty
identifier text")
+
+ try:
+ found = self.symbol_resolver.lookup_condition(name, self._section)
+ except SandboxDenialError:
+ raise
+ except SymbolResolutionError:
+ # Only an unqualified name can be a variable; a dotted or
qualified one keeps the resolver's error.
+ if '.' not in name and ':' not in name:
+ error = SymbolResolutionError(
+ "identifier", f"Undefined variable: '{name}'. Variables
must be declared in a VARS section.")
+ if suggestions :=
self.symbol_resolver.get_variable_suggestions(name, self._section):
+ error.add_symbol_suggestion(suggestions)
+ raise error from None
+ raise
+
+ self._drain_resolver_warnings(span)
+
+ if found.symbol is not None:
+ return ir.VarRef(symbol=found.symbol), False
+ return ir.BuiltinRef(key=found.key, entry=found.entry, arg=found.arg),
found.existence
+
+ def _assignment(self, node: ast.Assignment) -> ir.Stmt | None:
+ with self.trap(node.span):
+ value = self._operand(node.value, node.span)
+ value_text = ir.value_text(value)
+ add = node.operator == "+="
+ found = self.symbol_resolver.lookup_operator(node.name,
value_text, self._section, add=add)
+ self._drain_resolver_warnings(node.span)
+
+ if found.symbol is not None:
+ if isinstance(node.value, ast.IdentValue) and (rhs_symbol :=
self.symbol_resolver.symbol_for(node.value.raw)):
+ value = ir.VarRef(symbol=rhs_symbol)
+ return ir.SetVar(symbol=found.symbol, value=value,
span=node.span)
+
+ if add:
+ variant = ir.OpVariant.ADD
+ elif value_text == '""':
+ variant = ir.OpVariant.REMOVE
+ else:
+ variant = ir.OpVariant.SET
+
+ return ir.SetOp(
+ key=found.key, entry=found.entry, qualifier=found.qualifier,
value=value, variant=variant, span=node.span)
+ return None
+
+ def _statement_call(self, node: ast.FunctionCall) -> ir.Stmt | None:
+ with self.trap(node.span):
+ if '::' in node.name:
+ raise self.source_map.error_at(node.span, f"unknown procedure
'{node.name}': not loaded via 'use'")
+ args = [self._value(a, node.span) for a in node.args]
+ arg_texts = [ir.value_text(a) for a in args]
+ key, entry =
self.symbol_resolver.lookup_statement_function(node.name, arg_texts,
self._section)
+ self._drain_resolver_warnings(node.span)
+ return ir.CallOp(key=key, entry=entry,
args=tuple(map(self._normalized_arg, args, arg_texts)), span=node.span)
+ return None
+
+ @staticmethod
+ def _normalized_arg(arg: ir.RValue, text: str) -> ir.RValue:
+ """Validators normalize argument text in place; an argument they left
alone keeps its resolved form."""
+ if text == ir.value_text(arg):
+ return arg
+ return ir.RawArg(text=text)
+
+ @contextmanager
+ def _inlining(self, call: ast.FunctionCall):
+ sig = self.procs.get(call.name)
+ if call.name in self._proc_stack:
+ self._report(call.span, ValueError(f"circular procedure call: {'
-> '.join([*self._proc_stack, call.name])}"))
+ yield None
+ return
+
+ bindings = self._bind(sig, call)
+ if bindings is None:
+ yield None
+ return
+
+ body = tuple(self._substitute(node, bindings) for node in sig.body)
+ self._proc_stack.append(call.name)
+ try:
+ yield body
+ finally:
+ self._proc_stack.pop()
+
+ def _bind(self, sig: ProcEntry, call: ast.FunctionCall) -> dict[str, str]
| None:
+ with self.trap(call.span):
+ args = [self._binding_text(a, call.span) for a in call.args]
+ required = sum(1 for p in sig.params if p.default is None)
+ if not (required <= len(args) <= len(sig.params)):
+ expected = str(len(sig.params))
+ if required < len(sig.params):
+ expected = f"{required}-{len(sig.params)}"
+ raise ValueError(f"procedure '{sig.name}': expected {expected}
arg(s), got {len(args)}")
+
+ bindings = dict(zip((p.name for p in sig.params), args))
+ for param in sig.params[len(args):]:
+ bindings[param.name] = self._default_text(param.default,
call.span)
+ return bindings
+ return None
+
+ def _default_text(self, value: ast.ValueExpr, span: ast.Span) -> str:
+ """A string default loses its quotes but, unlike a call argument, is
not interpolated."""
+ if isinstance(value, ast.LiteralStringValue):
+ return value.raw
+ return self._binding_text(value, span)
+
+ def _binding_text(self, value, span: ast.Span) -> str:
+ """A binding is plain text: a string literal contributes its inner
text."""
+ if isinstance(value, ast.LiteralStringValue):
+ return ir.value_text(self._template(value.raw, span))[1:-1]
+ if isinstance(value, ast.ParamRef):
+ # _substitute replaced every bound `$param`, so one left here is
unbound.
+ raise self._unbound_param(value.raw, value.span or span)
+ return ir.value_text(self._value(value, span))
+
+ def _substitute(self, node: ast.BodyNode, bindings: dict[str, str]) ->
ast.BodyNode:
+
+ def sub_text(text: str) -> str:
+ return _PARAM_REF.sub(lambda m: bindings.get(m.group(1),
m.group(0)), text)
+
+ def sub_string(value):
+ if isinstance(value, ast.LiteralStringValue):
+ return ast.LiteralStringValue(raw=sub_text(value.raw))
+ return value
+
+ def sub_value(value):
+ if isinstance(value, ast.ParamRef) and value.raw in bindings:
+ # A bare `$param` takes its binding verbatim, so a string
argument reaches the operator unquoted.
+ return ast.IdentValue(raw=bindings[value.raw])
+ # An unbound `$param` stays a ParamRef so the statement using it
reports the error, in source order.
+ return sub_string(value)
+
+ # replace(), not a rebuild: a rebuilt node silently drops any field it
does not name,
+ # such as `else_span` or `modifier_span`, and the inlined copy then
diagnoses differently.
+ match node:
+ case ast.Assignment():
+ return replace(node, value=sub_value(node.value))
+ case ast.FunctionCall() if self.procs.get(node.name) is not None:
+ return replace(node, args=tuple(sub_value(a) for a in
node.args))
+ case ast.FunctionCall():
+ # An operator argument keeps its raw `$name` for the validator
to reject; only a
+ # string argument is substituted.
+ return replace(node, args=tuple(map(sub_string, node.args)))
+ case ast.IfBlock():
+ return replace(
+ node,
+ condition=self._substitute_cond(node.condition, sub_value),
+ body=tuple(self._substitute(b, bindings) for b in
node.body),
+ elif_branches=tuple(
+ replace(
+ arm,
+ condition=self._substitute_cond(arm.condition,
sub_value),
+ body=tuple(self._substitute(b, bindings) for b in
arm.body)) for arm in node.elif_branches),
+ else_body=tuple(self._substitute(b, bindings) for b in
node.else_body))
+ return node
+
+ def _substitute_cond(self, node: ast.ConditionExpr, sub_value: Callable)
-> ast.ConditionExpr:
+ match node:
+ case ast.Comparison():
+ return replace(node, right=sub_value(node.right))
+ case ast.LogicalOp():
+ return replace(
+ node, left=self._substitute_cond(node.left, sub_value),
right=self._substitute_cond(node.right, sub_value))
+ case ast.NotOp():
+ return replace(node,
operand=self._substitute_cond(node.operand, sub_value))
+ case ast.Group():
+ return replace(node, inner=self._substitute_cond(node.inner,
sub_value))
+ return node
+
+ def _value(self, node, span: ast.Span) -> ir.RValue:
+ match node:
+ case ast.NumberValue() | ast.IdentValue() | ast.BoolValue() |
ast.IPValue():
+ return ir.RawArg(text=node.raw)
+ case ast.LiteralStringValue():
+ return self._template(node.raw, span)
+ case ast.ParamRef():
+ # Unsubstituted: the entry's validator produces the diagnostic.
+ return ir.RawArg(text=f"${node.raw}")
+ case ast.RegexValue():
+ return ir.RegexLit(text=node.raw)
+ case ast.SetValue():
+ return ir.SetLit(text=node.raw)
+ case ast.IpRangeValue():
+ return ir.IpRangeLit(text=node.raw)
+ raise SymbolResolutionError("value", f"Unsupported value {node!r}")
+
+ def _operand(self, node, span: ast.Span) -> ir.RValue:
+ """An assignment or comparison right-hand side, where an unbound
`$name` is an error."""
+ if isinstance(node, ast.ParamRef):
+ raise self._unbound_param(node.raw, node.span or span)
+ return self._value(node, span)
+
+ def _source_text(self, node, span: ast.Span) -> str:
+ if isinstance(node, ast.LiteralStringValue):
+ return f'"{node.raw}"'
+ return ir.value_text(self._value(node, span))
+
+ def _template(self, inner: str, span: ast.Span) -> ir.StrTemplate:
+ parts: list[str | ir.BuiltinRef | ir.VarRef | ir.ModifiedRef] = []
+ pos = 0
+ for m in RegexPatterns.SUBSTITUTE_PATTERN.finditer(inner):
+ if m.start() > pos:
+ parts.append(inner[pos:m.start()])
+ parts.append(self._template_part(m, span, inner))
+ pos = m.end()
+ if pos < len(inner):
+ parts.append(inner[pos:])
+ return ir.StrTemplate(parts=tuple(parts))
+
+ def _template_part(self, m, span: ast.Span, inner: str) -> str |
ir.BuiltinRef | ir.VarRef | ir.ModifiedRef:
+ try:
+ return self._template_part_inner(m, span)
+ except SandboxDenialError as exc:
+ # Unwrapped, so the sandbox message is still set.
+ exc.add_note(self._interpolation_note(inner))
Review Comment:
The CST emitter takes this note from the literal before `$param`
substitution (`visitor.py:239`). Here `inner` has already been substituted, so
for
```
procedure syn::p($a) {
inbound.req.X-A = "{bogus} $a";
}
```
called as `syn::p("v")`, Sema's note reads `"{bogus} v"` and the CST
emitter's reads `"{bogus} $a"`. The parity test compares notes, so this case
would fail it. The CST text is also the better one for the user, since it's
what appears in their file.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
