Download bin/semantics.py from DBax127/nexa: direct link, hf CLI and curl.
- Browser
- Download file 237 kB
-
https://huggingface.co/DBax127/nexa/resolve/main/bin/semantics.py
- Command line
-
hf download hf://DBax127/nexa/bin/semantics.py
-
curl -L -o semantics.py https://huggingface.co/DBax127/nexa/resolve/main/bin/semantics.py
237 kB
| #!/usr/bin/env python3 | |
| """Semantic checking: parse the generated code instead of pattern-matching it. | |
| WHY THIS EXISTS | |
| --------------- | |
| A violation_pattern is a regex over generated code. It catches textual | |
| signatures, never meaning. Two of the corpus's critical rules were recorded in | |
| NO_PATTERN_REASON as unreachable for exactly that reason: | |
| odoo.constrains-must-list-every-field-read | |
| odoo.stored-compute-needs-complete-depends | |
| Both need the same thing: a set comparison between what a decorator declares and | |
| what the method body actually reads. That is a tree walk, not a regex, and | |
| Python's ast module makes it tractable without leaving the standard library. | |
| WHAT A CHECKER OWES | |
| ------------------- | |
| The same contract a violation_pattern owes, enforced by corpus.py validate: it | |
| must fire on its rule's own 'wrong' sample and stay silent on its 'correct' | |
| sample. A checker that catches nothing is useless; one that fires on good code | |
| trains you to ignore it, which is worse. | |
| HONEST SCOPE | |
| ------------ | |
| Python, and Odoo's XML data files, which expat reads with line numbers. There | |
| is no JavaScript or TypeScript parser in the standard library, so the TS/JS | |
| side gets balance_js -- a delimiter-balance scanner that is string-, comment- | |
| and template-aware. That is a syntax smoke test, not a parser, and the layer | |
| above reports UNKNOWN rather than PASS for everything it cannot see. | |
| Stdlib only. | |
| """ | |
| import ast | |
| import builtins | |
| import collections | |
| import importlib.util | |
| import re | |
| import textwrap | |
| import xml.parsers.expat | |
| import context | |
| CHECKER_VERSION = "1.0.0" | |
| # Attributes reached off a recordset that are ORM plumbing rather than fields. | |
| # Deliberately short: every name added here is a field a checker will stop | |
| # noticing, so the list earns entries only for names that can never be one. | |
| ORM_ATTRS = frozenset([ | |
| "env", "ids", "id", "pool", "sudo", "with_context", "with_user", "browse", | |
| "_name", "_table", "_cr", "_uid", "_context", "_origin", "_fields", | |
| "_description", "_inherit", "_inherits", "_order", "_rec_name", | |
| ]) | |
| # ------------------------------------------------------------------ parsing | |
| def parse_python(code): | |
| """Parse, then compile. Returns (tree, error) with at most one error set. | |
| Both steps run because they fail on different things. ast.parse rejects bad | |
| grammar; compile additionally rejects constructs that parse but cannot be | |
| assembled -- return outside a function, await outside async def, duplicate | |
| parameter names. Generated code hits the second class more often than the | |
| first, because a model emits a plausible fragment lifted out of the context | |
| that would have made it legal. | |
| """ | |
| try: | |
| tree = ast.parse(code) | |
| except SyntaxError as exc: | |
| return None, _syntax_error(exc) | |
| try: | |
| compile(tree, "<answer>", "exec") | |
| except (SyntaxError, ValueError, TypeError) as exc: | |
| return tree, _syntax_error(exc) | |
| return tree, None | |
| def _syntax_error(exc): | |
| return (getattr(exc, "lineno", None), getattr(exc, "msg", None) or str(exc)) | |
| Fragment = collections.namedtuple( | |
| "Fragment", "tree error notes source line_offset") | |
| # The four labels build_messages asks for. When they appear INSIDE the fence the | |
| # block is not broken code, it is code followed by prose, and calling that a | |
| # syntax error is the kind of false alarm that gets a checker switched off. | |
| _LABEL_RE = re.compile( | |
| r"^(Enforcement layer|Rules applied|Edge cases handled|Not handled)\s*:", re.I) | |
| def parse_fragment(code): | |
| """Parse generated code that may be a fragment lifted out of a class. | |
| WHY THIS IS NOT JUST parse_python: an Odoo answer is very often a method | |
| shown at the indentation it had inside its model class, and a bare ast.parse | |
| calls that 'unexpected indent'. Reporting a correct answer as uncompilable is | |
| the worst failure a syntax layer has available -- it is wrong, it is loud, | |
| and it masks every layer beneath it. | |
| Three repairs are attempted, cheapest first, and each one that works is | |
| RECORDED rather than hidden. The note is the finding: 'the labelled sections | |
| were inside the code fence' is a real thing to tell the user, it just is not | |
| a syntax error. | |
| line_offset is how many lines were added in front of the original code, so a | |
| caller reporting line numbers can subtract it and stay honest about where in | |
| the answer the problem is. | |
| """ | |
| tree, error = parse_python(code) | |
| if tree is not None and error is None: | |
| return Fragment(tree, None, [], code, 0) | |
| notes = [] | |
| candidate = code | |
| lines = candidate.splitlines() | |
| for index, line in enumerate(lines): | |
| if _LABEL_RE.match(line.strip()): | |
| candidate = "\n".join(lines[:index]).rstrip() | |
| notes.append("the labelled answer sections were inside the code fence") | |
| break | |
| if notes: | |
| tree, error = parse_python(candidate) | |
| if tree is not None and error is None: | |
| return Fragment(tree, None, notes, candidate, 0) | |
| dedented = textwrap.dedent(candidate) | |
| if dedented != candidate: | |
| tree, error = parse_python(dedented) | |
| if tree is not None and error is None: | |
| notes.append("the block is indented: a method shown inside its class body") | |
| return Fragment(tree, None, notes, dedented, 0) | |
| candidate = dedented | |
| # Ragged indentation dedent cannot fix -- a method whose decorator sits at | |
| # one level and whose body sits at another. Give it a class to live in. | |
| wrapped = "class _Fragment:\n" + "\n".join( | |
| (" " + line if line.strip() else line) for line in candidate.splitlines()) | |
| tree, wrapped_error = parse_python(wrapped) | |
| if tree is not None and wrapped_error is None: | |
| notes.append("the block only parses as a class body fragment") | |
| return Fragment(tree, None, notes, wrapped, 1) | |
| return Fragment(None, error or wrapped_error, notes, code, 0) | |
| # --------------------------------------------------------- path resolution | |
| def _attr_path(node, aliases): | |
| """Resolve an attribute chain to a dotted field path relative to self. | |
| rec.line_ids.price with rec bound to self yields ('line_ids', 'price'). | |
| Returns None when the base is not something known to be self-derived, which | |
| is the honest answer for a name the checker never saw bound. | |
| A chain may start from records picked by position. rec.line_ids[0].price and | |
| rec.line_ids.sorted("sequence")[:1].price both yield ('line_ids', 'price'), | |
| because indexing or slicing a recordset gives records of the same path. | |
| Stopping at the subscript left the price unread, and a compute that takes | |
| its value from the first line looked as if it read nothing from it. | |
| A chain may start from a call, too, when the call hands back records of the | |
| same path: rec.sudo().partner_id.name reads partner_id.name as surely as | |
| rec.partner_id.name does. Which calls those are is _resolve_iterable's | |
| list, the one a loop's iterable already goes through, so a loop and a read | |
| cannot disagree about it (ARCHITECTURE 3.78). | |
| """ | |
| segments = [] | |
| cursor = node | |
| while isinstance(cursor, ast.Attribute): | |
| segments.append(cursor.attr) | |
| cursor = cursor.value | |
| if isinstance(cursor, ast.Subscript) and _indexes_records(cursor): | |
| base = _resolve_iterable(cursor.value, aliases) | |
| elif isinstance(cursor, ast.Call): | |
| base = _resolve_iterable(cursor, aliases) | |
| elif isinstance(cursor, ast.Name): | |
| base = aliases.get(cursor.id) | |
| else: | |
| return None | |
| if base is None: | |
| return None | |
| return base + tuple(reversed(segments)) | |
| def _indexes_records(subscript): | |
| """Does this subscript pick records by position: x[0], x[-1], x[1:]? | |
| record[name] is the other thing a subscript on a recordset does, and it | |
| reads the field called name, not records of the same path. So a subscript is | |
| looked through only when its index is a literal integer or a slice. | |
| rec.line_ids[i] could be either, and stays unresolved. | |
| """ | |
| index = subscript.slice | |
| if isinstance(index, ast.Slice): | |
| return True | |
| if isinstance(index, ast.UnaryOp) and isinstance(index.op, ast.USub): | |
| index = index.operand | |
| return isinstance(index, ast.Constant) and type(index.value) is int | |
| def _aliases(fn): | |
| """Names that stand for a recordset, and the path each reaches it by. | |
| self is the root. 'for rec in self' makes rec the root as well. 'for line in | |
| rec.line_ids' makes line the path ('line_ids',), which is what turns a later | |
| line.price into the dependency line_ids.price. | |
| Resolved to a fixpoint rather than in source order, because a comprehension | |
| can bind a name from a loop variable introduced further down the tree and a | |
| single forward pass would miss the second hop. | |
| A name assigned records, pending = rec.review_ids.sorted("sequence")[:1], | |
| stays unbound, and so does a loop over one. Binding it when every value | |
| assigned to it resolves to one path is exact, and was measured: on odoo | |
| core the depends check's reads through such names read 37 of 113 true, | |
| 28 of them smart defaults, the same ratio as the lambda binding | |
| (ARCHITECTURE 3.72 and 3.80). Measured again on the tree facts, which | |
| decide 27 of the 113, it still adds more false findings than true (3.87). | |
| """ | |
| aliases = {"self": ()} | |
| for _ in range(6): | |
| before = len(aliases) | |
| for node in ast.walk(fn): | |
| if isinstance(node, (ast.For, ast.AsyncFor, ast.comprehension)): | |
| target, source = node.target, node.iter | |
| else: | |
| continue | |
| if not isinstance(target, ast.Name): | |
| continue # tuple unpacking: no single recordset to bind | |
| path = _resolve_iterable(source, aliases) | |
| if path is not None: | |
| aliases[target.id] = path | |
| if len(aliases) == before: | |
| break | |
| return aliases | |
| def _resolve_iterable(node, aliases): | |
| """The recordset path an expression iterates over, if it is one. | |
| Unwraps the calls that return a recordset of the same shape, and a subscript | |
| by position, which picks records out of one. mapped() is excluded on | |
| purpose: it returns values, not records, and treating it as a recordset | |
| would invent dependencies that do not exist. | |
| """ | |
| if isinstance(node, ast.Name): | |
| return aliases.get(node.id) | |
| if isinstance(node, ast.Attribute): | |
| return _attr_path(node, aliases) | |
| if isinstance(node, ast.Subscript) and _indexes_records(node): | |
| return _resolve_iterable(node.value, aliases) | |
| if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute): | |
| if node.func.attr in ("filtered", "sorted", "exists", "with_context", "sudo"): | |
| return _resolve_iterable(node.func.value, aliases) | |
| return None | |
| def _called_attrs(fn): | |
| """Attribute nodes sitting in a Call's function position. | |
| rec.write(...) is a method, not a field. Without this every ORM call would | |
| be reported as an undeclared field read, which is the fastest possible way | |
| to make a checker worth ignoring. | |
| """ | |
| return {id(node.func) for node in ast.walk(fn) | |
| if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute)} | |
| def _message_only_attrs(fn): | |
| """Attribute nodes reachable only from inside a raise statement. | |
| Found by crosscheck: the checker fired on the 'correct' sample of | |
| odoo.constrains-iterate-recordset, which reads rec.display_name to build a | |
| ValidationError message. A field read to phrase an error does not decide | |
| whether the invariant holds, so it cannot decide when the constraint must | |
| re-run either -- demanding it in @api.constrains would be noise. | |
| A field read in both the condition and the message is still caught, because | |
| only the nodes inside the raise are excluded, not the field name. | |
| A message assembled into a variable and raised on the next line is the same | |
| read for the same reason, and the inline form was all this covered. That | |
| missed the spelling OCA actually uses: | |
| msg = self.env._("... %(journal)s ...", journal=journal.display_name) | |
| raise ValidationError(msg) | |
| which reported display_name as a field the constraint depends on. | |
| Two more spellings came out of reading OpenSPP and odoo core on 2026-09-26. | |
| A log line is a message too: `_logger.warning("... %s", rec.name)` reads a | |
| field to phrase something, and five of OpenSPP's live false positives were | |
| exactly that. And a message can be collected before it is raised -- | |
| `sold_out.append(_("... %(name)s", name=event.name))`, then | |
| `raise ValidationError("\\n".join(sold_out))` -- which reads the same way as | |
| the assignment form above. | |
| """ | |
| excluded = set() | |
| raised = set() | |
| for node in ast.walk(fn): | |
| if isinstance(node, ast.Call) and _is_log_call(node): | |
| for inner in ast.walk(node): | |
| excluded.add(id(inner)) | |
| continue | |
| if not isinstance(node, ast.Raise): | |
| continue | |
| for inner in ast.walk(node): | |
| excluded.add(id(inner)) | |
| if isinstance(inner, ast.Name): | |
| raised.add(inner.id) | |
| if raised: | |
| for node in ast.walk(fn): | |
| if (isinstance(node, ast.Assign) and len(node.targets) == 1 | |
| and isinstance(node.targets[0], ast.Name) | |
| and node.targets[0].id in raised): | |
| for inner in ast.walk(node.value): | |
| excluded.add(id(inner)) | |
| elif (isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) | |
| and node.func.attr in ("append", "extend") | |
| and isinstance(node.func.value, ast.Name) | |
| and node.func.value.id in raised): | |
| for arg in node.args: | |
| for inner in ast.walk(arg): | |
| excluded.add(id(inner)) | |
| return excluded | |
| LOG_METHODS = frozenset(["debug", "info", "warning", "warn", "error", "exception", | |
| "critical", "log"]) | |
| def _is_log_call(call): | |
| """`_logger.warning(...)`, `logger.info(...)`, `logging.error(...)`.""" | |
| func = call.func | |
| if not (isinstance(func, ast.Attribute) and func.attr in LOG_METHODS): | |
| return False | |
| owner = func.value | |
| name = owner.id if isinstance(owner, ast.Name) else \ | |
| owner.attr if isinstance(owner, ast.Attribute) else "" | |
| return name == "logging" or name.lower().lstrip("_").endswith(("logger", "log")) | |
| def _field_reads(fn, aliases): | |
| """Every field path the body READS, as (path, lineno). | |
| Store context is excluded: rec.total = ... is the compute writing its | |
| result, and counting it would make every stored compute demand a dependency | |
| on the field it produces. | |
| """ | |
| skip = _called_attrs(fn) | _message_only_attrs(fn) | |
| reads = [] | |
| for node in ast.walk(fn): | |
| if not isinstance(node, ast.Attribute) or id(node) in skip: | |
| continue | |
| if not isinstance(node.ctx, ast.Load): | |
| continue | |
| path = _attr_path(node, aliases) | |
| # An Odoo field name never starts with an underscore -- that space is | |
| # reserved for ORM internals and mixin configuration, and a field there | |
| # would collide with them. ORM_ATTRS names the specific ones this code | |
| # meets; the prefix is the general rule behind the list. Without it, a | |
| # mixin reading self._time_window_overlap_check_field was told to add it | |
| # to @api.constrains, where the ORM ignores it. | |
| if path and path[0] not in ORM_ATTRS and not path[0].startswith("_"): | |
| reads.append((path, node.lineno)) | |
| return reads | |
| def _named_reads(fn, aliases): | |
| """Field paths a recordset method names in a string, as (path, lineno). | |
| account.invoice_ids.mapped("amount_residual") reads invoice_ids.amount_residual | |
| exactly as sum(inv.amount_residual for inv in account.invoice_ids) does, but | |
| the only attribute access in it is invoice_ids. Read by attributes alone, the | |
| first spelling looked fully declared by @api.depends("invoice_ids") and the | |
| second did not, so the checker was silent on the more idiomatic of the two. | |
| mapped() was the first of these (ARCHITECTURE 3.66). Its siblings name what | |
| they read the same way: filtered("a.b") tests a.b on every record, | |
| grouped("f") keys on f, sorted("f") orders on f, and filtered_domain() reads | |
| the field of every leaf written out in its domain. _named_paths says which | |
| paths each call names. | |
| Only literals are followed. mapped(name) stays unread: what a variable | |
| holds is a guess from here, and every guess in this checker is a finding | |
| somebody has to read. A lambda stays unread too, and that one is not a | |
| guess -- its parameter is each record of the receiver, exactly. It was | |
| measured instead, and what the lambdas in these trees test is mostly not a | |
| dependency (ARCHITECTURE 3.72, and again on the tree facts, 3.87). The | |
| receiver resolves the way a loop's iterable does, through filtered(), | |
| sorted(), sudo() and a subscript by position. A call inside a raise or a | |
| log line is phrasing a message, as an attribute read there is. | |
| """ | |
| quiet = _message_only_attrs(fn) | |
| reads = [] | |
| for node in ast.walk(fn): | |
| if not (isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute)) \ | |
| or id(node) in quiet: | |
| continue | |
| named = _named_paths(node) | |
| if not named: | |
| continue | |
| base = _resolve_iterable(node.func.value, aliases) | |
| if base is None: | |
| continue | |
| for segments in named: | |
| path = base + segments | |
| if path[0] not in ORM_ATTRS and not path[0].startswith("_"): | |
| reads.append((path, node.lineno)) | |
| return reads | |
| # Recordset methods whose one argument can be a field path. mapped("a.b") and | |
| # filtered("a.b") follow it through the relations; grouped("f") keys on f. | |
| PATH_ARGUMENT_METHODS = frozenset(["mapped", "filtered", "grouped"]) | |
| def _named_paths(call): | |
| """The field paths, relative to its receiver, that a recordset call names.""" | |
| method = call.func.attr | |
| if method in PATH_ARGUMENT_METHODS: | |
| if len(call.args) != 1 or call.keywords: | |
| return [] | |
| path = _literal_path(call.args[0]) | |
| return [path] if path else [] | |
| if method == "sorted": | |
| keys = list(call.args[:1]) + [k.value for k in call.keywords if k.arg == "key"] | |
| if len(keys) == 1 and isinstance(keys[0], ast.Constant) \ | |
| and isinstance(keys[0].value, str): | |
| return _order_fields(keys[0].value) | |
| return [] | |
| if method == "filtered_domain": | |
| if len(call.args) == 1 and not call.keywords \ | |
| and isinstance(call.args[0], (ast.List, ast.Tuple)): | |
| return _domain_fields(call.args[0]) | |
| return [] | |
| def _literal_path(node): | |
| """('a', 'b') for the string literal "a.b"; None for anything else.""" | |
| if not (isinstance(node, ast.Constant) and isinstance(node.value, str)): | |
| return None | |
| return _path_segments(node.value) | |
| def _path_segments(text): | |
| """('a', 'b') for "a.b"; None when the text is not a dotted field path.""" | |
| segments = tuple(text.strip().split(".")) | |
| return segments if all(segment.isidentifier() for segment in segments) else None | |
| ORDER_WORDS = frozenset(["asc", "desc", "nulls", "first", "last"]) | |
| def _order_fields(spec): | |
| """The field paths an order sorts on: "date desc, id" names date and id. | |
| Odoo 18 takes one field name here, as itemgetter(key). Odoo 19 takes an | |
| order: comma-separated terms, each a field with an optional .property and | |
| :granularity, then asc or desc and nulls first or last | |
| (_sorted_order_to_function in odoo/orm/models.py, 19.0). A spec with any | |
| other word in it is read not at all rather than in part. | |
| """ | |
| paths = [] | |
| for term in spec.split(","): | |
| words = term.split() | |
| if not words or any(word.lower() not in ORDER_WORDS for word in words[1:]): | |
| return [] | |
| path = _path_segments(words[0].split(":")[0]) | |
| if path is None: | |
| return [] | |
| paths.append(path) | |
| return paths | |
| def _domain_fields(domain): | |
| """The field path of every leaf a domain literal writes out. | |
| [("state", "=", "done"), ("partner_id.country_id", "in", ids)] names state and | |
| partner_id.country_id. A leaf whose operator is any or not any holds a | |
| domain on the related records, read through the leaf's own field: | |
| ("line_ids", "any", [("paid", "=", True)]) names line_ids and line_ids.paid. | |
| "&", "|" and "!" name nothing, and neither does a leaf built at run time. | |
| """ | |
| paths = [] | |
| for leaf in domain.elts: | |
| if not (isinstance(leaf, (ast.Tuple, ast.List)) and len(leaf.elts) == 3): | |
| continue | |
| field, operator, value = leaf.elts | |
| path = _literal_path(field) | |
| if path is None: | |
| continue | |
| paths.append(path) | |
| if isinstance(operator, ast.Constant) and operator.value in ("any", "not any") \ | |
| and isinstance(value, (ast.List, ast.Tuple)): | |
| paths.extend(path + inner for inner in _domain_fields(value)) | |
| return paths | |
| # A decorator that is present and cannot be read statically is neither absent | |
| # nor empty. @api.constrains(*FIELDS) used to collapse into "declares nothing", | |
| # which made every field the body reads a missing declaration and phrased it | |
| # "declares only nothing". | |
| UNREADABLE = object() | |
| def _note_unreadable(gaps, fn, decorator): | |
| """Record a method whose decorator could not be read. Not a finding. | |
| Nothing is wrong with @api.constrains(*FIELDS); it is simply outside what a | |
| static pass can decide. Reporting it as a violation is a false positive, and | |
| reporting nothing at all is the silence this corpus exists to avoid -- so it | |
| goes to the gap channel, where `check` prints it and no gate counts it. | |
| Once per method, however many checkers meet it. Two of them read | |
| @api.constrains, and the same unread method twice is one gap printed twice | |
| by `check` and listed twice by fieldtest. | |
| """ | |
| if gaps is not None: | |
| gap = ("{0}() has an @api.{1} whose arguments are not string literals, so " | |
| "what it declares could not be read -- that method was not " | |
| "checked".format(fn.name, decorator)) | |
| if gap not in gaps: | |
| gaps.append(gap) | |
| def _decorator(fn, module, name): | |
| """The @<module>.<name>(...) call decorating fn, or None. | |
| One definition of which decorator that is, so a finding anchored on its line | |
| and the arguments read from it cannot come from two different decorators. | |
| """ | |
| for dec in fn.decorator_list: | |
| if not isinstance(dec, ast.Call): | |
| continue | |
| func = dec.func | |
| if isinstance(func, ast.Attribute) and func.attr == name: | |
| if isinstance(func.value, ast.Name) and func.value.id == module: | |
| return dec | |
| elif isinstance(func, ast.Name) and func.id == name: | |
| return dec | |
| return None | |
| def _decorator_args(fn, module, name): | |
| """String literal arguments of @<module>.<name>(...). | |
| Three answers, not two: None when the decorator is absent, UNREADABLE when | |
| it is present and its arguments are not all string literals, and the list | |
| otherwise. An empty list stays an empty list -- @api.constrains() really | |
| does declare nothing, and that is readable. | |
| """ | |
| dec = _decorator(fn, module, name) | |
| if dec is None: | |
| return None | |
| if any(not (isinstance(a, ast.Constant) and isinstance(a.value, str)) | |
| for a in dec.args): | |
| return UNREADABLE | |
| return [a.value for a in dec.args] | |
| def _functions(tree): | |
| return [n for n in ast.walk(tree) | |
| if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef))] | |
| # ------------------------------------------------------------- the checkers | |
| def constrains_lists_every_field_read(code, gaps=None): | |
| """odoo.constrains-must-list-every-field-read | |
| @api.constrains fires only when a listed field appears in the write vals. A | |
| constraint that reads date_end while declaring only date_start never runs | |
| when date_end alone is written, and the invalid record is stored in silence. | |
| Only the first segment of each path is compared, because @api.constrains | |
| silently ignores dotted names -- declaring partner_id.country_id registers | |
| nothing, so a body reading it needs partner_id declared instead. That | |
| asymmetry with @api.depends is its own rule | |
| (odoo.constrains-ignores-dotted-names, constrains_declares_a_dotted_name | |
| below); this checker only respects it. | |
| Reads are the ones the depends check makes: attribute accesses, including | |
| one after a subscript by position, and the fields a recordset method names | |
| in a string (_named_reads). `if any(self.mapped("date_end"))` under | |
| @api.constrains("date_start") was silent. Measured before it was taken: OCA | |
| +1, OpenSPP 0, core +3, all read, 3 true. The false one filters on a | |
| non-stored compute, which @api.constrains cannot name: a class this rule | |
| already had for attribute reads (ARCHITECTURE 3.71). | |
| That class is answered now, for a field the constraint's own class | |
| declares. Odoo never runs a constraint for a field that is neither stored | |
| nor inversible, and warns "@constrains parameter ... is not writeable" when | |
| one is named, so telling the author to name it was advice that cannot be | |
| taken. What can trigger the check is a write of what the field is computed | |
| from: the roots of its compute's @api.depends, or the first field of its | |
| related path, followed through any of those that is unwriteable too | |
| (_unwriteable_inputs). A constraint that names them all is covered and | |
| reported nothing; one that does not is told which of them it is missing. | |
| When the inputs cannot be read, the finding says the field cannot be named | |
| rather than asking for it (ARCHITECTURE 3.82). | |
| """ | |
| tree, _ = parse_python(code) | |
| if tree is None: | |
| return [] | |
| findings = [] | |
| for fn in _functions(tree): | |
| declared = _decorator_args(fn, "api", "constrains") | |
| if declared is None: | |
| continue | |
| if declared is UNREADABLE: | |
| _note_unreadable(gaps, fn, "constrains") | |
| continue | |
| registered = {d.split(".")[0] for d in declared} | |
| missing = {} | |
| aliases = _aliases(fn) | |
| for path, lineno in _field_reads(fn, aliases) + _named_reads(fn, aliases): | |
| if path[0] not in registered: | |
| missing.setdefault(path[0], lineno) | |
| names = ", ".join(sorted(registered)) | |
| clause = "declares only {0}".format(names) if names else "declares nothing" | |
| owner = _owner_of(tree, fn) | |
| declarations = _class_fields(owner) if owner is not None else {} | |
| for field, lineno in sorted(missing.items()): | |
| if _writeable(declarations.get(field)) is not False: | |
| findings.append((lineno, | |
| "{0}() reads '{1}' but @api.constrains {2} -- " | |
| "writing '{1}' alone skips the check entirely".format( | |
| fn.name, field, clause))) | |
| continue | |
| inputs = _unwriteable_inputs(field, declarations, owner) | |
| if inputs is None: | |
| findings.append((lineno, | |
| "{0}() reads '{1}', which is not stored, so no " | |
| "write of '{1}' ever runs the check -- @api.constrains " | |
| "{2}, and needs the fields '{1}' is computed from " | |
| "instead".format(fn.name, field, clause))) | |
| continue | |
| lacking = sorted(inputs - registered) | |
| if not lacking: | |
| continue | |
| findings.append((lineno, | |
| "{0}() reads '{1}', which is not stored and is computed " | |
| "from {2}, but @api.constrains {3} -- writing {4} alone " | |
| "skips the check entirely".format( | |
| fn.name, field, ", ".join(sorted(inputs)), clause, | |
| " or ".join("'{0}'".format(name) for name in lacking)))) | |
| return findings | |
| def _class_fields(cls): | |
| """{field name: {keyword: value node}} for the fields a class body declares. | |
| Only `name = fields.Something(...)` statements directly in the body. A | |
| field declared in another module, or by a mixin, is not here, and a caller | |
| has to read that absence as unknown rather than as anything else. | |
| """ | |
| return {name: {k.arg: k.value for k in call.keywords if k.arg} | |
| for name, call in _class_field_calls(cls).items()} | |
| def _class_field_calls(cls): | |
| """{field name: the fields.Something(...) call declaring it}, as _class_fields.""" | |
| out = {} | |
| for statement in cls.body: | |
| if not (isinstance(statement, ast.Assign) and len(statement.targets) == 1 | |
| and isinstance(statement.targets[0], ast.Name) | |
| and isinstance(statement.value, ast.Call)): | |
| continue | |
| func = statement.value.func | |
| if not (isinstance(func, ast.Attribute) and isinstance(func.value, ast.Name) | |
| and func.value.id == "fields"): | |
| continue | |
| out[statement.targets[0].id] = statement.value | |
| return out | |
| def _writeable(declaration): | |
| """Can a write of this field run a constraint naming it? True, False or None. | |
| Odoo 18 runs a constraint when one of the names it declares is among the | |
| fields written, and a field that is neither stored nor inversible never is: | |
| _constraint_methods warns "@constrains parameter ... is not writeable" for | |
| it, and that is all (odoo/models.py, 18.0 at 3c3e3b3). From the declaration, | |
| per _get_attrs and setup_related in odoo/fields.py at the same commit: | |
| compute= not stored unless store=True; inversible with an inverse= | |
| related= not stored unless store=True; readonly unless readonly=False, | |
| and inversible only when it is not readonly | |
| neither a plain field, stored | |
| None when the declaration is absent or says it with something other than a | |
| literal: unknown, which is never read as not writeable. | |
| """ | |
| if declaration is None: | |
| return None | |
| if "compute" not in declaration and "related" not in declaration: | |
| return True | |
| store = declaration.get("store") | |
| if store is not None: | |
| if not isinstance(store, ast.Constant): | |
| return None | |
| if store.value: | |
| return True | |
| if "compute" in declaration: | |
| return "inverse" in declaration | |
| readonly = declaration.get("readonly") | |
| if readonly is None: | |
| return False | |
| if not isinstance(readonly, ast.Constant): | |
| return None | |
| return not readonly.value | |
| def _unwriteable_inputs(field, declarations, cls, seen=None): | |
| """The writeable fields an unwriteable field is computed from, or None. | |
| The roots of its compute's @api.depends, or the first field of its related | |
| path: @api.constrains names fields of its own model only, so a dependency | |
| through a relation is reached by naming the relation, as a read through one | |
| is. A root that is itself unwriteable is followed to what it is computed | |
| from. None when any step cannot be read: a compute method not in this class, | |
| @api.depends absent or not literal, a related path that is not a string. | |
| `id` never changes and is left out. | |
| """ | |
| seen = set(seen or ()) | {field} | |
| declaration = declarations[field] | |
| related = declaration.get("related") | |
| if related is not None: | |
| if not (isinstance(related, ast.Constant) and isinstance(related.value, str)): | |
| return None | |
| roots = {related.value.split(".")[0]} | |
| else: | |
| compute = declaration["compute"] | |
| if isinstance(compute, ast.Constant) and isinstance(compute.value, str): | |
| name = compute.value | |
| elif isinstance(compute, ast.Name): | |
| name = compute.id | |
| else: | |
| return None | |
| method = next((node for node in cls.body | |
| if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) | |
| and node.name == name), None) | |
| if method is None: | |
| return None | |
| depends = _decorator_args(method, "api", "depends") | |
| if depends is None or depends is UNREADABLE: | |
| return None | |
| roots = {path.split(".")[0] for path in depends} | |
| inputs = set() | |
| for root in roots - {"id"}: | |
| if root in seen: | |
| continue | |
| if _writeable(declarations.get(root)) is False: | |
| deeper = _unwriteable_inputs(root, declarations, cls, seen) | |
| if deeper is None: | |
| return None | |
| inputs |= deeper | |
| else: | |
| inputs.add(root) | |
| return inputs | |
| def constrains_declares_a_dotted_name(code, gaps=None): | |
| """odoo.constrains-ignores-dotted-names | |
| @api.constrains registers plain field names. A dotted one such as | |
| partner_id.country_id is accepted and the module installs, but it can never | |
| be among the written fields _validate_fields compares the declaration with, | |
| so the ORM never runs the check for it. Odoo 18 logs "@constrains parameter | |
| ... is not a field name" the first time the model collects its constraints, | |
| and that is all (odoo/models.py, 18.0 at 3c3e3b3). | |
| Was a regex for a quoted name of two segments in lowercase letters and | |
| underscores. partner_id.country_id.code has three, so where the regex wanted | |
| the closing quote there was a dot; and the class had no digits, so a name | |
| through any l10n_ field was invisible to it. _decorator_args reads the | |
| arguments exactly, and a computed list -- @api.constrains(*FIELDS), or the | |
| callable form Odoo also accepts -- goes to the gap channel rather than being | |
| guessed at. | |
| One finding per method, on the decorator line, where the names are and where | |
| a marker goes. The message says what is left to trigger the check: nothing | |
| when every name is dotted, and only the plain names when some are. | |
| 0 dotted names in 815 @api.constrains across odoo core, OCA and OpenSPP, so | |
| this moves nothing measured there. It is for code those trees do not hold, | |
| and for generated answers, where @api.depends beside it takes the same | |
| dotted name and means it. ARCHITECTURE 3.70. | |
| """ | |
| tree, _ = parse_python(code) | |
| if tree is None: | |
| return [] | |
| findings = [] | |
| for fn in _functions(tree): | |
| declared = _decorator_args(fn, "api", "constrains") | |
| if declared is None: | |
| continue | |
| if declared is UNREADABLE: | |
| _note_unreadable(gaps, fn, "constrains") | |
| continue | |
| dotted = [name for name in declared if "." in name] | |
| if not dotted: | |
| continue | |
| plain = [name for name in declared if "." not in name] | |
| effect = ("the ORM runs the check only when {0} is written".format( | |
| " or ".join(plain)) if plain else "the ORM never runs the check") | |
| findings.append(( | |
| _decorator(fn, "api", "constrains").lineno, | |
| "{0}() declares {1} in @api.constrains, which ignores dotted names " | |
| "-- {2}".format(fn.name, ", ".join("'{0}'".format(name) | |
| for name in dotted), effect))) | |
| return findings | |
| def _declared_covers(read, declared): | |
| """Is a field read already covered by something @api.depends declares? | |
| Exactly, or by a longer path that runs through it. @api.depends('a.b') | |
| registers the whole chain, so the ORM recomputes when a changes as well as | |
| when a.b does, and a body reading rec.a needs nothing further. Demanding the | |
| prefix be spelled out again reported correct code as stale: 42 of this | |
| checker's 104 hits on OpenSPP were exactly that, 40% of its output. | |
| The converse is the rule itself and must not follow: a declared 'a' does | |
| NOT cover a read of a.b, which is the stale value this exists to find. | |
| """ | |
| return any(d == read or d.startswith(read + ".") for d in declared) | |
| # Fields the ORM adds to every model itself, with a declaration this corpus | |
| # can read and a default of NOT stored. A model that wants one of these in the | |
| # database says so in its own file, and then the ordinary storedness test sees | |
| # it. Only display_name is here: the other automatic fields -- id, create_uid, | |
| # create_date, write_uid, write_date -- are never computed by module code, so | |
| # listing them would be describing a shape nobody writes. | |
| FRAMEWORK_UNSTORED = frozenset(["display_name"]) | |
| def depends_covers_compute_reads(code, facts=None, gaps=None): | |
| """odoo.stored-compute-needs-complete-depends | |
| A stored compute recomputes only when a declared dependency changes. A body | |
| reading line_ids.price while declaring only line_ids keeps a stale total in | |
| the database after a price edit, and every report built on that column is | |
| quietly wrong from then on. | |
| Full dotted paths are compared, not roots: Odoo needs line_ids.price spelled | |
| out, and line_ids alone does not imply it. | |
| That asymmetry runs one way only, and reading it as symmetric was this | |
| checker's largest source of false positives. A declared line_ids.price DOES | |
| cover a body that reads rec.line_ids -- the dependency is on the whole | |
| chain, so the ORM already recomputes when the relation itself changes. See | |
| _declared_covers. | |
| Two things are NOT reported, both learned by running this over 35k lines of | |
| OCA code rather than over the corpus's own samples: | |
| A path ending in .id. An id cannot change without the relation changing, | |
| so a read of journal_id.account_id.id is covered by journal_id.account_id, | |
| and one that is not declared is reported under that name. Odoo would take | |
| the .id spelling without a word: id is a field of every model, and | |
| resolve_depends keeps the relation it ends on and adds a trigger on id, | |
| which fires when a related record is deleted (odoo/models.py, | |
| odoo/fields.py, 18.0 at 3c3e3b3). Reporting it asked for a longer | |
| spelling of what was declared already. 13 of 121 findings on OCA were | |
| this. | |
| A read of a field this same method COMPUTES. Reading your own output | |
| mid-compute is usually a bug, but the fix is never "declare a dependency | |
| on it": Odoo drops a field from its own @api.depends without a word. | |
| resolve_depends skips it ("do not make self trigger itself", | |
| odoo/fields.py, 18.0 at 3c3e3b3), and every recompute and onchange runs | |
| off the triggers it builds, so the entry does nothing. Emitting advice | |
| that cannot be followed is worse than saying nothing, and there is no | |
| rule in the corpus about self-reads for this to become. 48 of 121 | |
| findings on OCA were reads of a computed field, most of them the | |
| method's own. | |
| A path through the output is exempt too, and there the premise is | |
| thinner. Odoo skips only the field's own name, so @api.depends("total.x") | |
| is kept, and the compute re-runs when x changes on the record total | |
| points to. Measured with the exemption narrowed to the bare field | |
| (ARCHITECTURE 3.73): OCA and OpenSPP 0, core +37, all read. 17 true, 16 | |
| of them a smart default checking its own earlier choice against | |
| configuration elsewhere; 20 false, where what total points to cannot | |
| change or write() refuses the change. So the exemption stays whole. | |
| What a method computes is what it assigns and what the file declares | |
| with compute="<that method>" (_computed_fields). A helper handed the | |
| records and the field name, update() and write() fill a field with no | |
| attribute store to see, and that field used to read as a dependency of | |
| the method producing it: calendar.recurrence's _compute_rrule on core, | |
| and the two OpenSPP findings already classified false as self-read. | |
| Core 2,358 -> 2,357, OpenSPP 163 -> 161, nothing added (ARCHITECTURE | |
| 3.72). | |
| And a third, added after classifying the survivors: a compute whose field is | |
| declared in this same file WITHOUT store=True. The rule is about STORED | |
| computes -- "written once and never recomputed... survives restarts" -- and | |
| none of that happens to a value the database never holds. 30 of 95 findings | |
| on OCA were non-stored computes, outside the rule's own scope. | |
| That test is deliberately conditional. Most of this checker's work is on | |
| generated snippets, where no field declaration is present at all -- the | |
| rule's own wrong sample is a bare @api.depends and a method body. So the | |
| file's evidence is used when it exists and the rule's assumption stands when | |
| it does not: proven-not-stored is dropped, unknown stays a finding. | |
| Reads are attribute accesses plus the fields a recordset method names in a | |
| string (_named_reads), and every exemption above applies to both. mapped() | |
| came first: sum(rec.invoice_ids.mapped("amount_residual")) under a bare | |
| @api.depends("invoice_ids") was silent while the same read spelled as a | |
| generator was reported. Core 2,336 -> 2,345; all 9 read, 8 true. filtered(), | |
| sorted(), grouped() and filtered_domain() followed, with a read after a | |
| subscript by position (rec.line_ids[0].amount, _attr_path): OCA +1, OpenSPP | |
| +6, core +17, all 24 read, 15 true. The 9 false are classes this checker | |
| already had for attribute reads (ARCHITECTURE 3.71). | |
| Three answers come from the tree when one was walked, each only where the | |
| module giving it is installed with this one (ARCHITECTURE 3.83): | |
| A compute whose fields every declaration in a complete walk leaves | |
| unstored is outside the rule, as one declared so in this file always | |
| was (_unstored_in_tree). | |
| Odoo merges @api.depends from every definition of the method in the | |
| model's MRO, so a dependency another definition declares is declared | |
| (_merge_into_declared). Only from the modules this one depends on: a | |
| sibling that overrides the same compute is not always installed. | |
| Odoo's triggers are transitive, through stored and unstored fields | |
| alike (get_field_trigger_tree, odoo/modules/registry.py at 3c3e3b3). | |
| A declared field that is computed or related covers what it depends on | |
| in turn (_covered_transitively). | |
| """ | |
| tree, _ = parse_python(code) | |
| if tree is None: | |
| return [] | |
| storedness = _compute_storedness(tree) | |
| computed = _computed_fields(tree) | |
| closure = context.module_closure(facts, (facts or {}).get("module")) | |
| findings = [] | |
| for fn in _functions(tree): | |
| declared = _decorator_args(fn, "api", "depends") | |
| if declared is None: | |
| continue | |
| if declared is UNREADABLE: | |
| _note_unreadable(gaps, fn, "depends") | |
| continue | |
| if storedness.get(fn.name) is False: | |
| continue | |
| owner = _owner_of(tree, fn) | |
| if owner is not None and _is_transient(owner): | |
| # A wizard's stored compute is written and read inside one | |
| # interaction. "Written once and never recomputed, survives | |
| # restarts, diverges from reality" is about a row that outlives the | |
| # thing it was computed from, and this one does not. | |
| continue | |
| model = _class_model(owner)[0] if owner is not None else None | |
| if storedness.get(fn.name) is None and _unstored_in_tree(facts, model, fn.name): | |
| continue | |
| registered = set(declared) | _merge_into_declared(facts, closure, model, fn.name) | |
| produced = _assigned_attrs(fn) | |
| if produced and produced <= FRAMEWORK_UNSTORED and \ | |
| storedness.get(fn.name) is None: | |
| # A field the framework adds to every model, which this file does | |
| # not redeclare. "Unknown storedness" is the right reading for an | |
| # ordinary field declared in another module; it is the wrong one | |
| # here, because the framework's own declaration is visible and | |
| # says not stored. odoo/models.py adds display_name as | |
| # `fields.Char(automatic=True, compute='_compute_display_name', | |
| # search=...)` with no store, so an override that omits a | |
| # dependency cannot leave a stale row: there is no row. | |
| # | |
| # 121 of the 1051 findings this rule made on odoo core were | |
| # _compute_display_name overrides of exactly that shape. The | |
| # narrowing costs nothing measurable: all 10 hand-verified true | |
| # positives that compute display_name -- every one of them in | |
| # OpenSPP -- declare `store=True` on it in their own file, so | |
| # storedness reads True for them and they are untouched. | |
| continue | |
| missing = {} | |
| aliases = _aliases(fn) | |
| own = produced | computed.get(fn.name, set()) | |
| for path, lineno in _field_reads(fn, aliases) + _named_reads(fn, aliases): | |
| while len(path) > 1 and path[-1] in ORM_ATTRS: | |
| path = path[:-1] | |
| if path[0] in own: | |
| continue | |
| dotted = ".".join(path) | |
| if _declared_covers(dotted, registered): | |
| continue | |
| if not _covered_transitively(facts, closure, model, registered, dotted): | |
| missing.setdefault(dotted, lineno) | |
| if missing: | |
| # `nexa check` reads the modules this one depends on only now, for a | |
| # read the module's own declarations leave uncovered; most files never | |
| # have one (ARCHITECTURE 3.86). More declarations can only cover more, | |
| # so the reads left are the only ones to ask about again. | |
| wider = context.with_dependencies(facts) | |
| if wider is not None: | |
| reach = context.module_closure(wider, wider.get("module")) | |
| merged = registered | _merge_into_declared(wider, reach, model, fn.name) | |
| missing = {dotted: lineno for dotted, lineno in missing.items() | |
| if not _declared_covers(dotted, merged) | |
| and not _covered_transitively(wider, reach, model, merged, dotted)} | |
| for dotted, lineno in sorted(missing.items()): | |
| findings.append((lineno, | |
| "{0}() reads '{1}' but @api.depends does not declare it -- " | |
| "the stored value goes stale when '{1}' changes".format( | |
| fn.name, dotted))) | |
| return findings | |
| def _class_model(cls): | |
| """(model, named, inherited) for a class, read as Odoo reads it. | |
| `_name`, or failing that a single `_inherit`. Odoo names a class that | |
| inherits several models without a `_name` after the class itself | |
| (odoo/models.py, 18.0 at 3c3e3b3); that is no model the tree declares, so | |
| it is None here. | |
| """ | |
| name = inherit = None | |
| for node in cls.body: | |
| if not (isinstance(node, ast.Assign) and len(node.targets) == 1 | |
| and isinstance(node.targets[0], ast.Name)): | |
| continue | |
| if node.targets[0].id == "_name": | |
| name = _literal_string(node.value) | |
| elif node.targets[0].id == "_inherit": | |
| inherit = _literal_strings(node.value) | |
| inherit = inherit or [] | |
| return (name or (inherit[0] if len(inherit) == 1 else None), name is not None, | |
| inherit) | |
| def _lineage(facts, model, closure): | |
| """The model and every model it inherits, transitively, by the inheritance | |
| the modules in `closure` declare, or every module's when closure is None.""" | |
| parents = facts.get("orm_parents") or {} | |
| seen, stack = {model}, [model] | |
| while stack: | |
| for module, parent in parents.get(stack.pop(), ()): | |
| if (closure is None or module in closure) and parent not in seen: | |
| seen.add(parent) | |
| stack.append(parent) | |
| return seen | |
| def _unstored_in_tree(facts, model, method): | |
| """Does every declaration the walk holds leave this compute's fields unstored? | |
| Only from a complete walk that names the model and everything it inherits: | |
| "nothing stores it" is true only of everything read, and an extension | |
| cannot see the declaration that created the field. Adding compute= to a | |
| field without store=True makes it unstored, whatever an earlier | |
| declaration of it said (_get_attrs, odoo/fields.py at 3c3e3b3), so one | |
| store=True anywhere is what keeps it stored. | |
| """ | |
| if not (facts and facts.get("walked") and facts.get("complete") and model): | |
| return False | |
| named = facts.get("orm_named") or set() | |
| lineage = _lineage(facts, model, None) | |
| if any(m not in named for m in lineage): | |
| return False | |
| declared = facts.get("orm_fields") or {} | |
| found = False | |
| for field in {f for m in lineage for f, rows in declared.get(m, {}).items() | |
| if any(d["compute"] == method for _, d in rows)}: | |
| found = True | |
| stores = [d["store"] for m in lineage for _, d in declared.get(m, {}).get(field, ())] | |
| if any(store is True or store == "?" for store in stores): | |
| return False | |
| return found | |
| def _merged_depends(facts, closure, model, method): | |
| """The @api.depends of every definition of `method` in the model's MRO | |
| that a module in `closure` declares.""" | |
| if not facts or closure is None or not model: | |
| return set() | |
| methods = facts.get("orm_methods") or {} | |
| merged = set() | |
| for m in _lineage(facts, model, closure): | |
| for module, depends in methods.get(m, {}).get(method, ()): | |
| if module in closure and depends: | |
| merged.update(depends) | |
| return merged | |
| def _merge_into_declared(facts, closure, model, method): | |
| """What Odoo adds to this method's own @api.depends from its other | |
| definitions: the depends check's use of _merged_depends.""" | |
| return _merged_depends(facts, closure, model, method) | |
| def _covered_transitively(facts, closure, model, declared, read): | |
| """Does a declared dependency reach `read` through what the fields on its | |
| way depend on? | |
| Each step of a declared path is a field of the model the path has reached | |
| so far. A computed one depends on its compute's merged @api.depends and a | |
| related one on its path, both relative to that model, so a step replaces | |
| the rest of the path with the field's own dependency and keeps the prefix | |
| that reached it. A relational field moves the next step to its comodel; a | |
| related one stops there, since its comodel is not in its declaration. Only | |
| declarations from modules in `closure` count. | |
| The search goes toward the read. A step keeps its prefix, so a step taken | |
| where the path has already left the read's can never come back to it, and | |
| is not taken. Every path met is then a prefix of the read followed by one | |
| dependency some field declares, a finite set, so the search ends without a | |
| limit, and its answer does not depend on the order it looks in. | |
| """ | |
| if not facts or closure is None or not model: | |
| return False | |
| target = tuple(read.split(".")) | |
| queue = collections.deque(sorted(tuple(path.split(".")) for path in declared)) | |
| seen = set(queue) | |
| while queue: | |
| path = queue.popleft() | |
| if path[:len(target)] == target: | |
| return True | |
| current = model | |
| for index in range(min(len(path), len(target))): | |
| rows = _declarations(facts, closure, current, path[index]) | |
| for row in rows: | |
| for dependency in _field_dependencies(facts, closure, current, row): | |
| reached = path[:index] + tuple(dependency.split(".")) | |
| if reached not in seen: | |
| seen.add(reached) | |
| queue.append(reached) | |
| if path[index] != target[index]: | |
| break | |
| comodel = next((row["comodel"] for row in rows if row["comodel"]), None) | |
| if comodel is None: | |
| break | |
| current = comodel | |
| return False | |
| def _declarations(facts, closure, model, field): | |
| """Every declaration of `field` on the model's lineage from a module in | |
| `closure`, in a fixed order.""" | |
| fields = facts.get("orm_fields") or {} | |
| return [row for m in sorted(_lineage(facts, model, closure)) | |
| for module, row in fields.get(m, {}).get(field, ()) | |
| if module in closure] | |
| def _field_dependencies(facts, closure, model, row): | |
| """What one declaration of a field makes it depend on, relative to its model.""" | |
| if row["related"]: | |
| return [row["related"]] | |
| if row["compute"]: | |
| return sorted(_merged_depends(facts, closure, model, row["compute"])) | |
| return [] | |
| def _test_atoms(fn): | |
| """Every condition tested by an if or while in this function, with and/or | |
| flattened. `a and (b or c)` yields a, b, c.""" | |
| atoms = [] | |
| stack = [node.test for node in ast.walk(fn) | |
| if isinstance(node, (ast.If, ast.While))] | |
| while stack: | |
| node = stack.pop() | |
| if isinstance(node, ast.BoolOp): | |
| stack.extend(node.values) | |
| else: | |
| atoms.append(node) | |
| return atoms | |
| def _is_literal_comparison(node): | |
| """A comparison of something against a literal, and nothing more. | |
| Deliberately narrow: this decides whether a constraint is doing ONLY what a | |
| CHECK can do, so anything it does not recognise has to count against the | |
| claim, never for it. | |
| """ | |
| if not isinstance(node, ast.Compare): | |
| return False | |
| operands = [node.left] + list(node.comparators) | |
| return any(isinstance(o, ast.Constant) and not isinstance(o.value, bool) | |
| and isinstance(o.value, (int, float, str)) for o in operands) | |
| def _compute_storedness(tree): | |
| """{compute method name: True|False} from the field declarations in scope. | |
| A name absent from this map is unknown, not unstored: the declaration may | |
| live in another file, or the code may be a fragment with no declaration at | |
| all. Callers must treat absence and False differently. | |
| """ | |
| stored = {} | |
| for node in ast.walk(tree): | |
| if not isinstance(node, ast.Call) or not isinstance(node.func, ast.Attribute): | |
| continue | |
| compute = store = None | |
| for keyword in node.keywords: | |
| if keyword.arg == "compute" and isinstance(keyword.value, ast.Constant): | |
| compute = keyword.value.value | |
| elif keyword.arg == "store" and isinstance(keyword.value, ast.Constant): | |
| store = keyword.value.value | |
| if isinstance(compute, str): | |
| # A field extended in two places counts as stored if either stores | |
| # it: one declaration that writes to the database is enough for the | |
| # value to go stale in it. | |
| stored[compute] = bool(store) or bool(stored.get(compute)) | |
| return stored | |
| def _computed_fields(tree): | |
| """{compute method name: the field names this file declares it computes}. | |
| `account_type = fields.Selection(compute="_compute_account_type", ...)` | |
| maps _compute_account_type to {"account_type"}. That is the method's output | |
| whether or not the body assigns it where _assigned_attrs can see: a helper | |
| handed the records and the field name, update({...}), write({...}) and | |
| rec[name] = ... all fill it without an attribute store. | |
| """ | |
| out = collections.defaultdict(set) | |
| for node in ast.walk(tree): | |
| if not (isinstance(node, ast.Assign) and len(node.targets) == 1 | |
| and isinstance(node.targets[0], ast.Name) | |
| and isinstance(node.value, ast.Call)): | |
| continue | |
| for keyword in node.value.keywords: | |
| if (keyword.arg == "compute" and isinstance(keyword.value, ast.Constant) | |
| and isinstance(keyword.value.value, str)): | |
| out[keyword.value.value].add(node.targets[0].id) | |
| return out | |
| def _assigned_attrs(fn): | |
| """Attribute names this function writes: rec.total = ... -> {'total'}. | |
| These are the compute's outputs. _field_reads already drops the assignment | |
| itself, but a body that writes rec.total and elsewhere READS rec.total.name | |
| would otherwise be told to depend on the field it is producing. | |
| """ | |
| names = set() | |
| for node in ast.walk(fn): | |
| targets = [] | |
| if isinstance(node, ast.Assign): | |
| targets = node.targets | |
| elif isinstance(node, (ast.AugAssign, ast.AnnAssign)): | |
| targets = [node.target] | |
| for target in targets: | |
| for inner in ast.walk(target): | |
| if isinstance(inner, ast.Attribute) and isinstance(inner.ctx, ast.Store): | |
| names.add(inner.attr) | |
| return names | |
| _SQL_COLUMNS = re.compile(r"\b(?:UNIQUE|CHECK)\s*\(([^)]*)\)", re.I) | |
| def _searches_self(call): | |
| """Is this search on THIS model, rather than on another one? | |
| `self.search(...)` and `self.sudo().search(...)` are uniqueness on this | |
| table, which is what a UNIQUE entry replaces. `self.env["other"].search(...)` | |
| is uniqueness across tables, which no constraint on this table can express. | |
| """ | |
| receiver = call.func.value | |
| while isinstance(receiver, ast.Call) and isinstance(receiver.func, ast.Attribute): | |
| receiver = receiver.func.value | |
| return isinstance(receiver, ast.Name) and receiver.id == "self" | |
| def _owner_of(tree, fn): | |
| """The ClassDef a function is defined in, or None for a module-level one.""" | |
| for node in ast.walk(tree): | |
| if isinstance(node, ast.ClassDef) and any(child is fn for child in node.body): | |
| return node | |
| return None | |
| def _columns_already_constrained(tree, series=None): | |
| """{class node id: column names a SQL constraint on that class already covers} | |
| Two spellings, and neither works everywhere. | |
| `_code_unique = models.Constraint("UNIQUE(code)", msg)` is the 19 one: odoo | |
| 18.0 core has no models.Constraint at all and declares _sql_constraints in | |
| 174 files. `_sql_constraints = [(name, "UNIQUE(code)", msg)]` is the classic | |
| one and is IGNORED from 19: the registry logs a warning at load and the | |
| constraint is never created. | |
| So on a module that pins 19 or later, a _sql_constraints block is not a | |
| column being spoken for -- it is a column that nobody is watching and | |
| everybody thinks somebody is. Reading it as satisfaction made this checker | |
| silent on exactly the file that carries the bug: OpenSPP's | |
| spp.cr.type.reason.document declared (cr_type_id, reason) that way, | |
| duplicates accumulated unnoticed from 19.0.3.0.0, and the fix (208d9758) | |
| re-declared it as models.Constraint. This checker said nothing either side | |
| of that commit. | |
| `series` unknown keeps the old reading. Firing on every 16, 17 and 18 module | |
| that did exactly the right thing is a worse trade than missing the 19 case, | |
| and `check` already reports the absent manifest as a gap. | |
| Column names are read out of the SQL text, which is crude and sufficient: | |
| the question is only whether this column is already spoken for. | |
| """ | |
| dead_legacy = _series_at_least(series, 19.0) | |
| out = {} | |
| for node in ast.walk(tree): | |
| if not isinstance(node, ast.ClassDef): | |
| continue | |
| columns = set() | |
| for statement in node.body: | |
| if not isinstance(statement, ast.Assign): | |
| continue | |
| if dead_legacy and _assigns_name(statement, "_sql_constraints"): | |
| continue | |
| for literal in ast.walk(statement): | |
| if isinstance(literal, ast.Constant) and isinstance(literal.value, str): | |
| for group in _SQL_COLUMNS.findall(literal.value): | |
| columns.update( | |
| re.findall(r"[A-Za-z_][A-Za-z0-9_]*", group)) | |
| out[id(node)] = columns | |
| return out | |
| def _assigns_name(statement, name): | |
| return any(isinstance(t, ast.Name) and t.id == name for t in statement.targets) | |
| def legacy_sql_constraints_on_19(code, facts=None): | |
| """odoo.sql-constraints-are-ignored-from-19 | |
| A class-body `_sql_constraints` assignment that Odoo 19 will not read. From | |
| 19 the registry logs one warning at load and creates nothing, so every | |
| UNIQUE and CHECK in the block is a constraint nobody is enforcing and | |
| everybody thinks somebody is. | |
| The incident is OpenSPP's 208d9758 (#394): spp.cr.type.reason.document | |
| declared (cr_type_id, reason) this way, duplicates were saved from | |
| 19.0.3.0.0, and the fix re-declared it as models.Constraint and added a | |
| lint of its own because nothing in CI had caught it. The rest of that tree | |
| had already moved, 72 files of models.Constraint to this one holdout. | |
| Which series, three answers: | |
| 19 or later reported. | |
| 16 to 18 silent. _sql_constraints is the ONLY spelling there -- | |
| odoo 18.0 core declares it in 174 files and has no | |
| models.Constraint -- so a finding would be advice that | |
| cannot be taken. OCA's twenty are all 18.0. | |
| unknown silent when a tree was walked and does not pin a series, | |
| which is odoo core itself: its manifests carry versions | |
| like "1.0", and 174 findings there would all be wrong. | |
| Reported, advisory, when NOTHING was walked -- a bare | |
| snippet, validate's own samples -- because "nobody looked" | |
| and "looked and it does not say" are different answers, | |
| and silence on the first would hide the rule entirely. | |
| An empty list declares nothing and is not reported. | |
| """ | |
| facts = facts or {} | |
| series = facts.get("odoo_series") | |
| if _series_at_least(series, 19.0): | |
| grade = None | |
| elif series is None and not facts.get("walked"): | |
| grade = UNPROVEN | |
| else: | |
| return [] | |
| tree, _ = parse_python(code) | |
| if tree is None: | |
| return [] | |
| findings = [] | |
| for cls in ast.walk(tree): | |
| if not isinstance(cls, ast.ClassDef): | |
| continue | |
| for statement in cls.body: | |
| if isinstance(statement, ast.Assign): | |
| targets, value = statement.targets, statement.value | |
| elif isinstance(statement, ast.AnnAssign) and statement.value is not None: | |
| targets, value = [statement.target], statement.value | |
| else: | |
| continue | |
| if not any(isinstance(t, ast.Name) and t.id == "_sql_constraints" | |
| for t in targets): | |
| continue | |
| if isinstance(value, (ast.List, ast.Tuple)) and not value.elts: | |
| continue | |
| findings.append(( | |
| statement.lineno, | |
| "{0} declares _sql_constraints, which Odoo 19 ignores{1}: the " | |
| "registry logs a warning and creates no constraint, so nothing " | |
| "enforces it -- declare each one as models.Constraint".format( | |
| cls.name, "" if grade is None else | |
| " (the series is unknown here; this holds from 19)"), | |
| grade)) | |
| return findings | |
| def _series_at_least(series, floor): | |
| """Is this module pinned to `floor` or later? False when nothing said.""" | |
| try: | |
| return float(series) >= floor | |
| except (TypeError, ValueError): | |
| return False | |
| def prefer_sql_constraint_applies(code, facts=None): | |
| """odoo.prefer-sql-constraint | |
| Fires on an @api.constrains that PostgreSQL could have enforced itself. | |
| Two shapes qualify, and deliberately nothing else: | |
| uniqueness the body calls search or search_count. That is a uniqueness | |
| test with a race inside it -- two concurrent transactions | |
| both find nothing, both insert, and the duplicate lands. | |
| UNIQUE does not race. | |
| bound the body does nothing but compare its own declared columns | |
| against literals. CHECK(qty > 0) holds for every writer | |
| forever, including the ones that never load the ORM. | |
| NOT covered, and this is now a settled decision rather than an omission. | |
| CHECK(date_end >= date_start) is perfectly good SQL, so the obvious reading | |
| is that recall was left on the table. Two things were weighed, and widening | |
| lost on both: | |
| It would fire on another rule's correct sample. The canonical two-field | |
| constraint in this corpus is odoo.constrains-must-list-every-field-read's | |
| own fix -- constrains('date_start', 'date_end') guarding date_end < | |
| date_start. A widened checker flags it, `nexa crosscheck` reports a | |
| standing hit forever, and the two rules appear to disagree when they do | |
| not. A correct sample that another checker condemns is a corpus defect; | |
| that is exactly why odoo.constrains-iterate-recordset's sample was | |
| rewritten rather than exempted. | |
| The suggested replacement is weaker than what it replaces. PostgreSQL | |
| treats a CHECK as satisfied when its expression evaluates to NULL, so | |
| CHECK(date_end >= date_start) enforces nothing on a row where either date | |
| is still empty -- which, for two optional columns, is most rows for most | |
| of their life. The Python constraint it would replace usually carries an | |
| explicit `if rec.date_start and rec.date_end` guard saying the same thing | |
| on purpose. Advising the swap without that caveat reads as a | |
| tightening and is not one. A single-column bound does not have the same | |
| problem in practice: a numeric column with a default is rarely NULL. | |
| So the gap stays, pinned by | |
| tests/test_semantics.py::test_two_column_comparison_is_silent. Reopening it | |
| means answering both points, not just observing that the SQL is valid. | |
| Out of scope by definition: anything reaching a related record, calling a | |
| method, or touching self.env. A CHECK genuinely cannot express those, which | |
| is what odoo.check-constraint-cannot-cross-tables exists to say. | |
| This replaced a regex matching only the search_count shape, which meant the | |
| rule claimed ground its own checker never patrolled. | |
| TWO NARROWINGS, BOTH FROM `nexa recall --aim` | |
| --------------------------------------------- | |
| Placed against the method each fix edited, this rule scored 0 on-target and | |
| 47 other-symbol out of 49 -- the worst signal in the corpus by that measure. | |
| Reading the 85 findings on OpenSPP gave two classes where the advice cannot | |
| be followed at all, which is worse than advice nobody wanted: | |
| The search is on ANOTHER model. `self.env["other.model"].search(...)` | |
| tests uniqueness across tables, and a UNIQUE on this one cannot express | |
| that. `endpoint_route_handler._check_route_unique_across_models` is the | |
| clearest case: it loops over every consumer model in the registry. That | |
| belongs to check-constraint-cannot-cross-tables, which already says so. | |
| The class ALREADY declares the constraint being recommended. A model | |
| carrying `models.Constraint("UNIQUE(code)", ...)` or an `_sql_constraints` | |
| entry for the same column has taken the advice; the Python check beside it | |
| is belt and braces, and telling the author to do the thing they did is how | |
| a checker gets switched off. Both spellings are read, because Odoo 17 | |
| introduced the second and OpenSPP uses both. | |
| """ | |
| tree, _ = parse_python(code) | |
| if tree is None: | |
| return [] | |
| guarded = _columns_already_constrained( | |
| tree, (facts or {}).get("odoo_series")) | |
| findings = [] | |
| for fn in _functions(tree): | |
| declared = _decorator_args(fn, "api", "constrains") | |
| if declared is None or declared is UNREADABLE: | |
| continue | |
| roots = {d.split(".")[0] for d in declared} | |
| if roots and roots <= guarded.get(id(_owner_of(tree, fn)), set()): | |
| continue # the class already declares this constraint | |
| searching = [n for n in ast.walk(fn) | |
| if isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute) | |
| and n.func.attr in ("search", "search_count", "search_read") | |
| and _searches_self(n)] | |
| if searching: | |
| findings.append((searching[0].lineno, | |
| "{0}() tests uniqueness in Python -- two concurrent " | |
| "transactions can both pass it and both insert; a UNIQUE " | |
| "models.Constraint cannot race".format(fn.name))) | |
| continue | |
| if any(isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute) | |
| and n.func.attr in ("search", "search_count", "search_read") | |
| for n in ast.walk(fn)): | |
| continue # searches another model: not this rule's | |
| # Anything the database cannot see for itself disqualifies the whole | |
| # function. Two exemptions, both necessary: calls inside the raise, | |
| # because ValidationError(...) is how the finding gets reported rather | |
| # than part of what is being decided, and the decorator list, because | |
| # @api.constrains(...) is itself a Call and walking the whole function | |
| # node would let every constraint disqualify itself. | |
| in_raise = _message_only_attrs(fn) | |
| body = [node for stmt in fn.body for node in ast.walk(stmt)] | |
| if any(isinstance(n, ast.Call) and id(n) not in in_raise for n in body): | |
| continue | |
| reads = _field_reads(fn, _aliases(fn)) | |
| if not reads or any(len(path) != 1 or path[0] not in roots | |
| for path, _ in reads): | |
| continue | |
| # Every condition has to BE a literal comparison, not merely contain | |
| # one. `if record.storage == "url" and not record.url:` is a guard plus | |
| # a presence check, and only the guard is an ast.Compare -- so the old | |
| # test saw a column compared to a literal, concluded the function did | |
| # nothing else, and said so. That shape is five of the eight hits this | |
| # checker produced across two OCA repositories. | |
| if any(not _is_literal_comparison(atom) for atom in _test_atoms(fn)): | |
| continue | |
| for node in ast.walk(fn): | |
| if not isinstance(node, ast.Compare): | |
| continue | |
| operands = [node.left] + list(node.comparators) | |
| if not any(isinstance(o, ast.Constant) | |
| and isinstance(o.value, (int, float, str)) | |
| and not isinstance(o.value, bool) for o in operands): | |
| continue | |
| findings.append((node.lineno, | |
| "{0}() compares its own declared column against a literal " | |
| "and does nothing else -- the same rule as a CHECK in " | |
| "models.Constraint holds for every writer, including " | |
| "imports, XML-RPC and sudo".format(fn.name))) | |
| break | |
| return findings | |
| def ondelete_leaves_a_dangling_read(code, facts=None): | |
| """odoo.ondelete-is-a-decision | |
| Fires on a Many2one that declares no ondelete AND is read through somewhere | |
| in the same module. | |
| The ORM default for an OPTIONAL Many2one is 'set null'. Delete the parent and | |
| the reference is nulled; read it afterwards and Odoo hands back an empty | |
| recordset, whose attributes evaluate to False rather than raising. That is | |
| the rule's whole complaint -- "dangling references that later reads treat as | |
| empty" -- and it needs a read to happen at all. | |
| Two exemptions, both because the default is already correct there: | |
| required=True Odoo defaults a required Many2one to 'restrict', which is | |
| what this rule recommends. 56 of the 198 hits on OCA were | |
| fields already doing the right thing. | |
| related= a related field mirrors another record's value and takes | |
| its deletion behaviour from the source. | |
| no column a compute= without store=True is never written, so there | |
| is no foreign key for ondelete to govern. | |
| TransientModel a wizard's record is vacuumed. The harm here is a nulled | |
| reference read long afterwards, which needs a row that | |
| lives long enough for the parent to be deleted under it. | |
| NOT flagged, by choice: a Many2one declared and never dereferenced. Nulling a | |
| reference nobody reads through changes no answer, and flagging it produced | |
| 4.1 findings per 1000 lines of reviewed OCA code -- enough to bury the other | |
| nine Odoo rules under one that is a policy rather than a defect. | |
| That makes the checker narrower than the rule, which still asks for an | |
| explicit ondelete on every Many2one. The gap is deliberate and is the same | |
| one required=True already had. | |
| """ | |
| tree, _ = parse_python(code) | |
| if tree is None: | |
| return [] | |
| declared = _many2one_without_ondelete(tree) | |
| if not declared: | |
| return [] | |
| findings = [] | |
| for name, (lineno, comodel) in sorted(declared.items(), | |
| key=lambda kv: kv[1][0]): | |
| reads = _dereferences_of(tree, name) | |
| if not reads: | |
| continue | |
| trigger = context.parent_is_deleted(facts, comodel) | |
| if trigger == context.ABSENT: | |
| # The tree defines this parent and never deletes it. No delete, no | |
| # nulled reference, no dangling read -- the rule's entire complaint | |
| # is conditional on a thing that does not happen here. | |
| continue | |
| message = ("{0} declares no ondelete, so the ORM defaults it to " | |
| "'set null' -- and {0}.{1} on line {2} reads through it, " | |
| "which yields False rather than raising once the parent " | |
| "is deleted".format(name, reads[0][0], reads[0][1])) | |
| if trigger == context.PRESENT: | |
| message += (", and this tree does delete {0}".format(comodel)) | |
| findings.append((lineno, message, | |
| "defect" if trigger == context.PRESENT else None)) | |
| return findings | |
| def _many2one_without_ondelete(tree): | |
| """{field name: (line, comodel)} for every fields.Many2one(...) that leaves | |
| deletion behaviour to the ORM. required=True and related= are exempt. | |
| The comodel travels with the declaration because the trigger is a question | |
| about the PARENT -- whether anything deletes it -- and by the time the | |
| finding is written the call node is gone.""" | |
| found = {} | |
| transient = _transient_spans(tree) | |
| for node in ast.walk(tree): | |
| if not isinstance(node, ast.Assign) or len(node.targets) != 1: | |
| continue | |
| target = node.targets[0] | |
| if not isinstance(target, ast.Name): | |
| continue | |
| call = node.value | |
| if not (isinstance(call, ast.Call) and isinstance(call.func, ast.Attribute) | |
| and call.func.attr == "Many2one"): | |
| continue | |
| if _within(transient, node.lineno) or _has_no_stored_column(call): | |
| continue | |
| exempt = False | |
| for keyword in call.keywords: | |
| if keyword.arg in ("ondelete", "related"): | |
| exempt = True | |
| elif keyword.arg == "required" and _is_true(keyword.value): | |
| exempt = True | |
| if not exempt: | |
| found[target.id] = (node.lineno, _comodel_of(call)) | |
| return found | |
| def _is_true(node): | |
| return isinstance(node, ast.Constant) and node.value is True | |
| def _is_sql_view(cls): | |
| """True for a model declaring `_auto = False`: a SQL view, not a table. | |
| Odoo builds no table for one and therefore no foreign key, so `ondelete` | |
| is not a decision anybody can make about it -- there is no constraint for | |
| a value to be chosen for. Two of the 15 sampled ondelete findings on odoo | |
| core were report models of exactly this kind, hr.contract.history and | |
| purchase.bill.line.match. | |
| """ | |
| for node in cls.body: | |
| if not isinstance(node, ast.Assign): | |
| continue | |
| if any(isinstance(t, ast.Name) and t.id == "_auto" for t in node.targets): | |
| return isinstance(node.value, ast.Constant) and node.value.value is False | |
| return False | |
| def _transient_spans(tree): | |
| """Line ranges of every class this rule family does not apply to. | |
| A wizard's record is vacuumed, so the failure both ondelete-is-a-decision | |
| and stored-compute-needs-complete-depends describe -- a value written once | |
| and read wrong much later -- has no time in which to happen. A `_auto = | |
| False` model has no table at all, which rules out the same failure for a | |
| different reason: see _is_sql_view. Returned as spans because the | |
| declarations and methods they gate are found by walking the whole tree, | |
| which loses the class on the way down. | |
| check_company_missing_on_company_owned_relation already had this exemption | |
| and iterates classes directly; these two do not, hence the spans. | |
| """ | |
| return [(n.lineno, getattr(n, "end_lineno", n.lineno)) | |
| for n in ast.walk(tree) | |
| if isinstance(n, ast.ClassDef) and (_is_transient(n) or _is_sql_view(n))] | |
| def _within(spans, lineno): | |
| return any(start <= lineno <= end for start, end in spans) | |
| def _has_no_stored_column(call): | |
| """Does this field declaration put nothing in the database? | |
| compute= without store=True is calculated on read and never written, so | |
| there is no column for ondelete to act on and nothing for check_company to | |
| constrain. Both rules are about what the database HOLDS, and advice about a | |
| column that does not exist cannot be taken. | |
| """ | |
| computed = stored = False | |
| for keyword in call.keywords: | |
| if keyword.arg == "compute": | |
| computed = True | |
| elif keyword.arg == "store": | |
| stored = _is_true(keyword.value) | |
| return computed and not stored | |
| def _dereferences_of(tree, field): | |
| """[(attribute, line)] for every UNGUARDED <x>.<field>.<attribute> in the tree. | |
| The outer attribute is what makes it a read THROUGH the relation rather than | |
| a read OF it. `rec.partner_id` alone is a recordset test and is falsy-safe; | |
| `rec.partner_id.name` is the one that quietly becomes False. | |
| Reads inside a function that tests the field for truthiness are left out. | |
| The rule's complaint is that a nulled reference "yields False rather than | |
| raising", and an author who wrote `if not self.model_id: return` has already | |
| answered it -- the reference may dangle and nothing reads through it when it | |
| does. Measured on OpenSPP, that guard is the majority idiom, and flagging it | |
| made the difference between a rule about a defect and a rule about a | |
| construct. | |
| Function-level, deliberately. The dominant shape is a guard at the top of a | |
| method covering everything below it, and following the value any further | |
| than that is dataflow analysis this does not have. | |
| A read is placed by the field's name, not by the model it is read on. | |
| `mapped("<field>.<attribute>")` is a read too and is left out: measured on | |
| the three trees it added one finding, a quant's location read in | |
| stock.location's file, and five of the thirteen such reads were another | |
| model's field of the same name (ARCHITECTURE 3.79). | |
| """ | |
| guarded = set() | |
| for fn in _functions(tree): | |
| if _tests_for_emptiness(fn, field): | |
| guarded.update(id(node) for node in ast.walk(fn)) | |
| reads = [] | |
| for node in ast.walk(tree): | |
| if not isinstance(node, ast.Attribute) or id(node) in guarded: | |
| continue | |
| inner = node.value | |
| if not (isinstance(inner, ast.Attribute) and inner.attr == field): | |
| continue | |
| if node.attr in ORM_ATTRS: | |
| continue | |
| reads.append((node.attr, node.lineno)) | |
| return sorted(reads, key=lambda r: r[1]) | |
| def _tests_for_emptiness(fn, field): | |
| """Does this function check whether <x>.<field> is set, anywhere in it? | |
| `if not self.model_id: return`, `if rec.partner_id:`, `rec.partner_id and | |
| rec.partner_id.name`, and the ternary spelling all count. What is looked for | |
| is the field appearing as a CONDITION rather than as a value, which is the | |
| difference between asking whether it is there and assuming it is. | |
| """ | |
| def names_the_field(node): | |
| while isinstance(node, ast.UnaryOp) and isinstance(node.op, ast.Not): | |
| node = node.operand | |
| return isinstance(node, ast.Attribute) and node.attr == field | |
| for node in ast.walk(fn): | |
| tests = [] | |
| if isinstance(node, (ast.If, ast.While, ast.IfExp)): | |
| tests.append(node.test) | |
| elif isinstance(node, ast.BoolOp): | |
| tests.extend(node.values) | |
| elif isinstance(node, ast.Assert): | |
| tests.append(node.test) | |
| elif isinstance(node, ast.comprehension): | |
| tests.extend(node.ifs) | |
| for test in tests: | |
| if names_the_field(test): | |
| return True | |
| if isinstance(test, ast.BoolOp) and any( | |
| names_the_field(value) for value in test.values): | |
| return True | |
| return False | |
| # Core models that carry a company_id of their own, so a relation to one can | |
| # point across a tenant boundary. Curated rather than derived: Nexa has no Odoo | |
| # registry to ask, and a guessed list would fire on every Many2one in the corpus. | |
| COMPANY_OWNED = frozenset([ | |
| "account.journal", "account.account", "account.payment.term", | |
| "account.analytic.account", "stock.warehouse", "stock.location", | |
| "stock.picking.type", "product.pricelist", "res.partner.bank", | |
| ]) | |
| # Models that ARE the company, or are scoped to one by construction. A | |
| # check_company on a field of res.company has nothing to compare against; Odoo's | |
| # own fields.py carries a special branch for exactly that case. | |
| NOT_COMPANY_SCOPED = frozenset(["res.company", "res.users"]) | |
| def check_company_missing_on_company_owned_relation(code, facts=None): | |
| """odoo.check-company-on-relations | |
| Fires on a Many2one or Many2many to a company-owned model, declared without | |
| check_company, in a model that is itself company-scoped. | |
| All three conditions matter, and the regex that preceded this could express | |
| only the first two. Firing without the third is what produced 17 of its 37 | |
| hits on OCA: | |
| six were on res.company itself, where check_company has nothing to compare | |
| against -- odoo/fields.py has a branch for that case; | |
| eleven were TransientModel wizards. A wizard's records do not persist, and | |
| the harm this rule names is a multi-company client seeing another tenant's | |
| data on their DOCUMENT. Flagging wizards is not wrong so much as it is a | |
| different, much smaller problem wearing a critical rule's severity. | |
| "Company-scoped" is read from the code, not guessed: the class declares a | |
| company_id field of its own. A class that only inherits one cannot be | |
| resolved without an Odoo registry, so it is skipped rather than assumed -- | |
| silence over a false critical. | |
| A module-level declaration with no class around it is treated as one | |
| implicit model body. That is the shape a generated answer arrives in, and | |
| real modules always put fields in a class, so it costs nothing on real code. | |
| """ | |
| trigger = context.multi_company_in_use(facts) | |
| if trigger == context.ABSENT: | |
| # One company cannot leak to another. The rule is about a boundary this | |
| # codebase does not have, and every finding it could make here would be | |
| # about a harm with no mechanism. | |
| return [] | |
| tree, _ = parse_python(code) | |
| if tree is None: | |
| return [] | |
| classes = [n for n in ast.walk(tree) if isinstance(n, ast.ClassDef)] | |
| scopes = classes if classes else [tree] | |
| findings = [] | |
| for scope in scopes: | |
| if isinstance(scope, ast.ClassDef): | |
| if _model_names(scope) & NOT_COMPANY_SCOPED: | |
| continue | |
| if _is_transient(scope) or _is_sql_view(scope): | |
| # A SQL view has no table, so no foreign key for check_company | |
| # to add a guard to: the rows come from the query. OpenSPP's | |
| # fund_report was the one live false positive left for this rule. | |
| continue | |
| if not _declares_company_id(scope): | |
| continue | |
| for node in ast.walk(scope): | |
| if not (isinstance(node, ast.Call) | |
| and isinstance(node.func, ast.Attribute) | |
| and node.func.attr in ("Many2one", "Many2many")): | |
| continue | |
| comodel = _comodel_of(node) | |
| if comodel not in COMPANY_OWNED: | |
| continue | |
| if any(k.arg == "check_company" for k in node.keywords): | |
| continue | |
| if any(k.arg == "related" for k in node.keywords): | |
| # A mirror of a field on another record. The constraint belongs | |
| # on the source, and the source is flagged in its own right -- | |
| # so this was also reporting the same decision twice. | |
| continue | |
| if _has_no_stored_column(node): | |
| # No column, nothing to constrain. | |
| continue | |
| message = ("relation to {0} with no check_company, on a model " | |
| "that has its own company_id -- a record in one " | |
| "company can reference another company's {0}" | |
| .format(comodel)) | |
| if trigger == context.PRESENT: | |
| message += ", and this codebase runs multi-company" | |
| findings.append((node.lineno, message, | |
| "defect" if trigger == context.PRESENT else None)) | |
| return findings | |
| def _model_names(cls): | |
| """Every model name the class claims: _name and _inherit, string forms.""" | |
| names = set() | |
| for node in cls.body: | |
| if not isinstance(node, ast.Assign) or len(node.targets) != 1: | |
| continue | |
| target = node.targets[0] | |
| if not (isinstance(target, ast.Name) and target.id in ("_name", "_inherit")): | |
| continue | |
| value = node.value | |
| if isinstance(value, ast.Constant) and isinstance(value.value, str): | |
| names.add(value.value) | |
| elif isinstance(value, (ast.List, ast.Tuple)): | |
| for item in value.elts: | |
| if isinstance(item, ast.Constant) and isinstance(item.value, str): | |
| names.add(item.value) | |
| return names | |
| def _is_transient(cls): | |
| for base in cls.bases: | |
| if isinstance(base, ast.Attribute) and base.attr == "TransientModel": | |
| return True | |
| if isinstance(base, ast.Name) and base.id == "TransientModel": | |
| return True | |
| return False | |
| def _declares_company_id(scope): | |
| body = scope.body if isinstance(scope, (ast.ClassDef, ast.Module)) else [] | |
| for node in body: | |
| if not isinstance(node, ast.Assign) or len(node.targets) != 1: | |
| continue | |
| target = node.targets[0] | |
| if isinstance(target, ast.Name) and target.id == "company_id": | |
| return True | |
| return False | |
| def _comodel_of(call): | |
| """The model a relational field points at, positional or keyword.""" | |
| if call.args and isinstance(call.args[0], ast.Constant) \ | |
| and isinstance(call.args[0].value, str): | |
| return call.args[0].value | |
| for keyword in call.keywords: | |
| if keyword.arg == "comodel_name" and isinstance(keyword.value, ast.Constant): | |
| return keyword.value.value | |
| return None | |
| # ---------------------------------------------------- the hierarchy checker | |
| # Either spelling clears the rule. _check_recursion still works on 18.0 -- it | |
| # warns and delegates to _has_cycle -- so code carrying it is out of date, not | |
| # unguarded. | |
| CYCLE_GUARDS = frozenset(["_has_cycle", "_check_recursion"]) | |
| def parent_can_form_a_cycle(code): | |
| """odoo.recursion-check-on-hierarchy | |
| Fires on a `parent_id` Many2one that points at its own model, in a model | |
| nothing protects from a cycle. | |
| Was a regex for `parent_id = fields.Many2one`, cleared by either guard's | |
| name anywhere in the file. It could not ask the two questions the harm | |
| depends on, and both produced false positives that were classified true: | |
| Does the field point at its own model? A `parent_id` to a different | |
| model is an ordinary foreign key and cannot close a loop. OCA's | |
| account.cash.deposit.line.parent_id points at account.cash.deposit. | |
| Does the framework already refuse the cycle? A model with | |
| `_parent_store = True` keeps parent_path, and _parent_store_update | |
| raises "Recursion Detected." on any write that would make a record its | |
| own ancestor (odoo/models.py, present since 2019). Six of the seven | |
| verdicts once marked true -- account.asset.group on OCA, and spp.area, | |
| spp.area.type, spp.audit.rule, spp.consent.purpose and | |
| spp.hazard.category on OpenSPP -- were models doing exactly this. | |
| The one that survived both questions is the shape the rule is for: | |
| spp.dms.directory declares a parent_path CHAR but never sets | |
| _parent_store, so nothing maintains it and nothing refuses a loop, while | |
| its complete_name and root_directory_id computes recurse up the chain. | |
| Self-reference is read through the three spellings real modules use: a | |
| string, the class's own `_name` variable, and a module constant both are | |
| assigned from (`_name = _type_model`). A comodel that cannot be resolved | |
| keeps the old reading and fires, and so does a module-level declaration | |
| with no class around it -- the shape a generated answer arrives in. | |
| """ | |
| tree, _ = parse_python(code) | |
| if tree is None: | |
| return [] | |
| for node in ast.walk(tree): | |
| name = node.attr if isinstance(node, ast.Attribute) else \ | |
| node.id if isinstance(node, ast.Name) else None | |
| if name in CYCLE_GUARDS: | |
| return [] | |
| constants = _module_string_constants(tree) | |
| classes = [n for n in ast.walk(tree) if isinstance(n, ast.ClassDef)] | |
| findings = [] | |
| for scope in classes or [tree]: | |
| own = None | |
| if isinstance(scope, ast.ClassDef): | |
| if _parent_store_guards_parent_id(scope) or _is_sql_view(scope): | |
| # A view is written by nobody: its parent comes from the query | |
| # over the tables underneath, which is where a guard belongs. | |
| continue | |
| own = _own_model_name(scope, constants) | |
| for node in scope.body: | |
| if not (isinstance(node, ast.Assign) and len(node.targets) == 1 | |
| and isinstance(node.targets[0], ast.Name) | |
| and node.targets[0].id == "parent_id" | |
| and isinstance(node.value, ast.Call) | |
| and isinstance(node.value.func, ast.Attribute) | |
| and node.value.func.attr == "Many2one"): | |
| continue | |
| if not node.value.args and not any(k.arg == "comodel_name" | |
| for k in node.value.keywords): | |
| # `parent_id = fields.Many2one(tracking=3)` names no model: it | |
| # re-declares attributes of a field declared elsewhere, and the | |
| # relation, with any guard on it, lives with that declaration. | |
| continue | |
| comodel = _resolved_comodel(node.value, constants, own) | |
| if own is not None and comodel is not None and comodel != own: | |
| continue | |
| findings.append((node.lineno, | |
| "parent_id points at {0} itself and nothing refuses a " | |
| "cycle -- a record made its own ancestor loops every " | |
| "recursive read until the worker is killed. " | |
| "_parent_store = True would refuse it, and so would an " | |
| "@api.constrains calling _has_cycle()".format( | |
| comodel or own or "its own model"))) | |
| return findings | |
| def _module_string_constants(tree): | |
| """NAME = "string" at module level: how a model name gets shared.""" | |
| found = {} | |
| for node in tree.body: | |
| if (isinstance(node, ast.Assign) and len(node.targets) == 1 | |
| and isinstance(node.targets[0], ast.Name) | |
| and isinstance(node.value, ast.Constant) | |
| and isinstance(node.value.value, str)): | |
| found[node.targets[0].id] = node.value.value | |
| return found | |
| def _class_assignment(cls, attr): | |
| """The value a class-body assignment gives `attr`, chained ones included. | |
| Odoo's own test models write `_name = _description = 'x'`; reading only | |
| single-target assignments left those classes nameless, so a parent_id to | |
| a different model read as unresolvable and fired. | |
| """ | |
| for node in cls.body: | |
| if isinstance(node, ast.Assign) and any( | |
| isinstance(t, ast.Name) and t.id == attr for t in node.targets): | |
| return node.value | |
| return None | |
| def _own_model_name(cls, constants): | |
| """The model this class defines or extends, when one name says it.""" | |
| value = _class_assignment(cls, "_name") or _class_assignment(cls, "_inherit") | |
| if isinstance(value, ast.Constant) and isinstance(value.value, str): | |
| return value.value | |
| if isinstance(value, ast.Name): | |
| return constants.get(value.id) | |
| return None | |
| def _parent_store_guards_parent_id(cls): | |
| """True when the ORM itself refuses a cycle through parent_id.""" | |
| store = _class_assignment(cls, "_parent_store") | |
| if not _is_true(store): | |
| return False | |
| # _parent_store maintains the field _parent_name names, parent_id unless | |
| # said otherwise. A model storing some other chain leaves parent_id bare. | |
| name = _class_assignment(cls, "_parent_name") | |
| return name is None or (isinstance(name, ast.Constant) and name.value == "parent_id") | |
| def _resolved_comodel(call, constants, own): | |
| literal = _comodel_of(call) | |
| if literal is not None: | |
| return literal | |
| expressions = [call.args[0]] if call.args else [] | |
| expressions += [k.value for k in call.keywords if k.arg == "comodel_name"] | |
| for expr in expressions: | |
| if isinstance(expr, ast.Name): | |
| if expr.id == "_name": | |
| return own | |
| if expr.id in constants: | |
| return constants[expr.id] | |
| return None | |
| # ------------------------------------------------------ the XML data checker | |
| # Models whose records an administrator tunes through the UI after install, so | |
| # that an upgrade rewriting them undoes a decision somebody made on purpose. | |
| # | |
| # Chosen by measurement, not by opinion. Odoo core puts its own records of each | |
| # under noupdate almost without exception -- ir.config_parameter 26 of 28, | |
| # ir.sequence 16 of 17, decimal.precision 7 of 8, and stages 30 of 34 across | |
| # seven stage models -- while views, actions and reports, which must upgrade, | |
| # sit at 0 to 2%. ir.cron is the exception to that method: core protects only | |
| # 39 of 93. It came in on an incident: OpenSPP's #473 is two crons and two | |
| # parameters in one file with no noupdate, and names the harm in its title -- | |
| # an admin who changed an interval or disabled a cron is reset by the next | |
| # upgrade. #479 fixed it with noupdate and a migration. Reading every cron | |
| # finding across the three trees then narrowed it to one shape: see | |
| # editable_record_without_noupdate. | |
| # | |
| # ir.rule is out on purpose, although core protects 566 of 598. A record rule | |
| # frozen by noupdate never receives the security fix its module ships later, | |
| # which is a trade-off a checker has no business deciding for anybody. | |
| EDITABLE_DATA_MODELS = frozenset([ | |
| "ir.config_parameter", "ir.sequence", "decimal.precision", "ir.cron", | |
| ]) | |
| def _is_editable_data_model(model): | |
| return model in EDITABLE_DATA_MODELS or (model or "").endswith(".stage") | |
| def _is_false_literal(text): | |
| return text.strip().lower() in ("false", "0") | |
| def parse_xml(code): | |
| """(True, None), or (False, (line, message)) when expat cannot read it.""" | |
| parser = xml.parsers.expat.ParserCreate() | |
| try: | |
| parser.Parse(code.encode("utf-8"), True) | |
| except xml.parsers.expat.ExpatError as exc: | |
| return False, (exc.lineno, xml.parsers.expat.errors.messages.get( | |
| exc.code, "not well-formed")) | |
| return True, None | |
| # An Odoo data document. An answer from the model can carry one inside prose | |
| # and Python, where the whole text is not XML; the documents inside it still are. | |
| _XML_DOCUMENT = re.compile(r"<(odoo|openerp)\b.*?</\1\s*>", re.S) | |
| def _xml_documents(code): | |
| """(line offset, text) for each Odoo data document the code holds.""" | |
| ok, _ = parse_xml(code) | |
| if ok: | |
| return [(0, code)] | |
| return [(code.count("\n", 0, m.start()), m.group(0)) | |
| for m in _XML_DOCUMENT.finditer(code)] | |
| def _records_outside_noupdate(document): | |
| """(line, model, xml id, ships disabled) for each <record> no noupdate="1" | |
| reaches. | |
| noupdate is inherited from the enclosing <odoo>, <openerp> or <data>, and | |
| an inner element can switch it back off, so it is tracked per element. | |
| `ships disabled` is whether the record sets its `active` field false. | |
| """ | |
| stack, found = [], [] | |
| current = {} | |
| parser = xml.parsers.expat.ParserCreate() | |
| def start(tag, attrs): | |
| inherited = stack[-1] if stack else False | |
| if attrs.get("noupdate") is not None: | |
| inherited = attrs["noupdate"].strip().lower() in ("1", "true") | |
| stack.append(inherited) | |
| if tag == "record" and not inherited: | |
| current.clear() | |
| current.update(line=parser.CurrentLineNumber, model=attrs.get("model"), | |
| id=attrs.get("id"), disabled=False, text=None) | |
| elif tag == "field" and current and attrs.get("name") == "active": | |
| if attrs.get("eval") is not None: | |
| current["disabled"] = _is_false_literal(attrs["eval"]) | |
| else: | |
| current["text"] = [] | |
| def chars(data): | |
| if current.get("text") is not None: | |
| current["text"].append(data) | |
| def end(tag): | |
| stack.pop() | |
| if tag == "field" and current.get("text") is not None: | |
| current["disabled"] = _is_false_literal("".join(current["text"])) | |
| current["text"] = None | |
| elif tag == "record" and current: | |
| found.append((current["line"], current["model"], current["id"], | |
| current["disabled"])) | |
| current.clear() | |
| parser.StartElementHandler, parser.EndElementHandler = start, end | |
| parser.CharacterDataHandler = chars | |
| try: | |
| parser.Parse(document.encode("utf-8"), True) | |
| except xml.parsers.expat.ExpatError: | |
| return [] | |
| return found | |
| def editable_record_without_noupdate(code): | |
| """odoo.noupdate-on-user-editable-data | |
| Fires on each <record> of a model administrators tune (see | |
| EDITABLE_DATA_MODELS) that no noupdate="1" reaches, at the record's own | |
| line. | |
| Was a regex for `<odoo>` immediately followed by `<record`, cleared by the | |
| word noupdate anywhere in the file. So it fired once, on <odoo>, for any | |
| record at all -- a view, a menu, an action, all of which have to upgrade -- | |
| and one noupdate block anywhere silenced every record outside it. OpenSPP's | |
| spp_starter_sp_mis config_parameters.xml after its own fix still carries a | |
| parameter outside the block, and the regex called the file clean. | |
| It had also never been fired at real code: fieldtest and recall read .py | |
| files only, and this rule reads XML. Its `unproven` grade was a statement | |
| about where the tools looked. | |
| """ | |
| findings = [] | |
| for offset, document in _xml_documents(code): | |
| for line, model, xid, disabled in _records_outside_noupdate(document): | |
| if not _is_editable_data_model(model): | |
| continue | |
| if model == "ir.cron" and not disabled: | |
| # A cron's schedule is usually the mechanism, not a setting: | |
| # 57 of 62 cron findings across the three trees read false, and | |
| # core keeps 54 of its own 93 crons updatable on purpose. The | |
| # shape that held up ships disabled for an operator to switch | |
| # on -- which is exactly the decision an upgrade then undoes. | |
| continue | |
| findings.append((offset + line, | |
| "{0} record '{1}' has no noupdate=\"1\" -- every module " | |
| "upgrade rewrites it with the shipped value, silently " | |
| "undoing what an administrator changed. Adding noupdate " | |
| "now does not reach databases that already have it: " | |
| "ir_model_data.noupdate also needs flipping in a " | |
| "migration".format(model, xid or "?"))) | |
| return findings | |
| # ------------------------------------------------------- the cursor checkers | |
| # The names an Odoo database cursor is reached by. self.env.cr and self._cr are | |
| # the two the framework provides; `cr` alone is the argument name every | |
| # migration script and post-init hook uses. Anything else called .execute() or | |
| # .commit() is somebody else's object and none of our business. | |
| CURSOR_NAMES = frozenset(["cr", "_cr", "cursor", "_cursor"]) | |
| def _cursor_call(node, method): | |
| """True when this Call is <something>.cr.<method>(...). | |
| The attribute immediately before the method name is what decides it, so | |
| self.env.cr.execute, self._cr.execute and a bare cr.execute all match while | |
| connection.execute and session.commit do not. | |
| """ | |
| if not isinstance(node, ast.Call): | |
| return False | |
| func = node.func | |
| if not isinstance(func, ast.Attribute) or func.attr != method: | |
| return False | |
| owner = func.value | |
| if isinstance(owner, ast.Attribute): | |
| return owner.attr in CURSOR_NAMES | |
| if isinstance(owner, ast.Name): | |
| return owner.id in CURSOR_NAMES | |
| return False | |
| def _root_name(node): | |
| """The name a dotted chain starts from. self.env.cr -> 'self'.""" | |
| while isinstance(node, ast.Attribute): | |
| node = node.value | |
| return node.id if isinstance(node, ast.Name) else None | |
| def _independent_cursors(scope): | |
| """Names in this scope standing for a cursor the caller does not own. | |
| `with Registry(...).cursor() as new_cr` opens a SECOND connection, and an | |
| Environment built on it commits its own transaction. That is not the | |
| caller's unit of work being split -- it is the documented way to write a | |
| batch job that must not hold one transaction open for the whole run, and | |
| the rule's own instruction is "let the caller own the transaction", which | |
| this already does. | |
| Bound POSITIVELY, by watching the cursor be opened, rather than by treating | |
| anything not spelled self.env.cr as independent. The two field instances of | |
| this rule are exactly the two sides of that line: OpenSPP's expiry cron | |
| commits self.env.cr inside its loop and is a true positive its author marked | |
| and kept, while OCA's batch_unlink commits a cursor it opened three lines | |
| earlier and is not. | |
| """ | |
| names = set() | |
| for _ in range(3): | |
| before = len(names) | |
| for node in ast.walk(scope): | |
| if isinstance(node, ast.withitem): | |
| call = node.context_expr | |
| if (isinstance(call, ast.Call) | |
| and isinstance(call.func, ast.Attribute) | |
| and call.func.attr == "cursor" | |
| and isinstance(node.optional_vars, ast.Name)): | |
| names.add(node.optional_vars.id) | |
| elif (isinstance(node, ast.Assign) and len(node.targets) == 1 | |
| and isinstance(node.targets[0], ast.Name) | |
| and isinstance(node.value, ast.Call)): | |
| # api.Environment(new_cr, ...) carries the new cursor forward. | |
| if any(isinstance(a, ast.Name) and a.id in names | |
| for a in node.value.args): | |
| names.add(node.targets[0].id) | |
| if len(names) == before: | |
| break | |
| return names | |
| def commit_inside_a_loop(code): | |
| """odoo.commit-inside-a-loop-leaves-half-written-data | |
| A commit ends the transaction the ORM is working inside. Done once per | |
| iteration, the loop stops being one unit of work: iteration seven raising | |
| leaves iterations one to six already written, the caller's except clause | |
| has nothing left to roll back, and the user sees an error over data that | |
| stayed. | |
| Only loops, not every commit. A commit in a cron batch or a migration | |
| script is a deliberate decision about batch size, and a rule that fired on | |
| all of them would be argued with rather than read. Inside a loop over a | |
| recordset it is almost never deliberate -- it is somebody flushing as they | |
| go -- and that is the shape this reports. | |
| The receiver decides, not the method name: `.commit()` on anything that is | |
| not a cursor belongs to some other library. | |
| """ | |
| tree, _ = parse_python(code) | |
| if tree is None: | |
| return [] | |
| # Which names mean an independent cursor, per function. Scoped rather than | |
| # gathered over the file, so a `new_cr` opened in one method cannot excuse a | |
| # commit in another that happens to reuse the name. | |
| spans = [(fn.lineno, getattr(fn, "end_lineno", fn.lineno), | |
| _independent_cursors(fn)) for fn in _functions(tree)] | |
| def independent_at(lineno): | |
| best = None | |
| for start, end, names in spans: | |
| if start <= lineno <= end and (best is None or start > best[0]): | |
| best = (start, names) | |
| return best[1] if best else set() | |
| findings = [] | |
| for loop in ast.walk(tree): | |
| if not isinstance(loop, (ast.For, ast.AsyncFor, ast.While)): | |
| continue | |
| # The loop's own body, not its else clause: an else runs once, after | |
| # the loop, and a commit there is a commit after a loop. | |
| for statement in loop.body: | |
| for node in ast.walk(statement): | |
| if not _cursor_call(node, "commit"): | |
| continue | |
| if _root_name(node.func.value) in independent_at(node.lineno): | |
| continue | |
| findings.append(( | |
| node.lineno, | |
| "cr.commit() inside a loop -- an exception on a later " | |
| "iteration leaves the earlier ones written, and the " | |
| "caller cannot roll the batch back")) | |
| return sorted(set(findings)) | |
| def sql_built_by_string_formatting(code): | |
| """odoo.sql-needs-query-parameters | |
| cr.execute("... id = %s", (rec.id,)) is correct and contains a percent | |
| sign. cr.execute("... id = %s" % rec.id) is the bug and contains the same | |
| percent sign. A regex cannot tell them apart, which is why this is an AST | |
| pass: the difference is whether the substitution happens in Python before | |
| the cursor sees the string, and that is a BinOp rather than a character. | |
| Four spellings, all of them the same mistake: | |
| "..." % value BinOp with Mod | |
| "...".format(value) a call to .format on a string | |
| f"...{value}..." JoinedStr | |
| "..." + value BinOp with Add, one side a string | |
| Only the FIRST argument is examined. The second is the parameter tuple and | |
| building that with formatting is somebody's business, not a SQL injection. | |
| """ | |
| tree, _ = parse_python(code) | |
| if tree is None: | |
| return [] | |
| findings = [] | |
| for node in ast.walk(tree): | |
| if not _cursor_call(node, "execute") or not node.args: | |
| continue | |
| how = _how_the_query_was_built(node.args[0]) | |
| if how: | |
| findings.append(( | |
| node.args[0].lineno, | |
| "the query handed to cr.execute() is built by {0} -- pass the " | |
| "values as the second argument so the driver quotes them".format(how))) | |
| return sorted(set(findings)) | |
| def _how_the_query_was_built(node): | |
| """The name of the string operation that produced this query, or "".""" | |
| if isinstance(node, ast.JoinedStr): | |
| return "an f-string" | |
| if isinstance(node, ast.BinOp): | |
| if isinstance(node.op, ast.Mod): | |
| return "percent formatting" | |
| if isinstance(node.op, ast.Add) and (_is_a_string(node.left) | |
| or _is_a_string(node.right)): | |
| return "string concatenation" | |
| if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) \ | |
| and node.func.attr == "format" and not _is_composed_sql(node.func.value): | |
| return "str.format()" | |
| return "" | |
| def _is_composed_sql(node): | |
| """psycopg2's sql.SQL("... {}").format(sql.Identifier(name)) is the CORRECT | |
| way to parameterise a table or column name, and it spells .format(). | |
| Found by running this over 229,600 lines of OpenSPP: the one finding that | |
| was wrong rather than merely suppressed was this construct, on a DROP INDEX | |
| whose identifier the surrounding code had already validated. A checker that | |
| tells an author to stop using the safe API is worse than one that says | |
| nothing. | |
| """ | |
| if not isinstance(node, ast.Call): | |
| return False | |
| func = node.func | |
| if isinstance(func, ast.Attribute): | |
| return func.attr in ("SQL", "Identifier", "Composed", "Literal") | |
| return isinstance(func, ast.Name) and func.id in ("SQL", "Identifier") | |
| def _is_a_string(node): | |
| return ((isinstance(node, ast.Constant) and isinstance(node.value, str)) | |
| or isinstance(node, ast.JoinedStr)) | |
| # Odoo's exception hierarchy, read out of odoo/exceptions.py and identical in | |
| # 16.0 through 18.0. Every one of these subclasses UserError, which is why | |
| # catching UserError first makes the specific handler below it dead code. | |
| ODOO_EXCEPTION_PARENT = { | |
| "AccessDenied": "UserError", | |
| "AccessError": "UserError", | |
| "MissingError": "UserError", | |
| "ValidationError": "UserError", | |
| "UserError": "Exception", | |
| "RedirectWarning": "Exception", | |
| } | |
| def _is_subclass_of(inner, outer): | |
| """Would `except outer` catch something raised as `inner`? | |
| Odoo's hierarchy comes from the table above; everything else is asked of | |
| the interpreter, so builtins stay correct without a second list to | |
| maintain. An unknown name answers False -- a class this pass has never | |
| heard of is not evidence of anything. | |
| """ | |
| if inner == outer: | |
| return True | |
| seen = set() | |
| cursor = inner | |
| while cursor in ODOO_EXCEPTION_PARENT and cursor not in seen: | |
| seen.add(cursor) | |
| cursor = ODOO_EXCEPTION_PARENT[cursor] | |
| if cursor == outer: | |
| return True | |
| if outer in ("BaseException", "Exception") and ( | |
| inner in ODOO_EXCEPTION_PARENT or _builtin_exception(inner)): | |
| return outer == "BaseException" or inner != "BaseException" | |
| first, second = _builtin_exception(inner), _builtin_exception(outer) | |
| return bool(first and second and issubclass(first, second)) | |
| def _builtin_exception(name): | |
| found = getattr(builtins, name, None) | |
| return found if isinstance(found, type) and issubclass(found, BaseException) else None | |
| def _handler_names(handler): | |
| """The exception names one `except` clause catches, tuples included.""" | |
| node = handler.type | |
| if node is None: | |
| return [] # a bare except, which Python forces last | |
| parts = node.elts if isinstance(node, ast.Tuple) else [node] | |
| names = [] | |
| for part in parts: | |
| if isinstance(part, ast.Name): | |
| names.append(part.id) | |
| elif isinstance(part, ast.Attribute): | |
| names.append(part.attr) # odoo.exceptions.AccessError | |
| return names | |
| def handler_order_shadows_a_subclass(code): | |
| """odoo.except-order-follows-the-hierarchy | |
| Python tries handlers in order and takes the first that matches, so a base | |
| class placed above a subclass makes the lower handler dead code. Nothing | |
| warns: it parses, it imports, it runs, and the specific branch is simply | |
| never entered. AccessDenied, AccessError, MissingError and ValidationError | |
| all subclass UserError -- read out of odoo/exceptions.py in 16.0 and 18.0, | |
| where the hierarchy is identical. Builtins are asked of the interpreter | |
| rather than listed, so `except Exception` above `except KeyError` is the | |
| same finding for the same reason. | |
| HOW THIS RULE CAME ABOUT, INCLUDING THE PART THAT WAS WRONG | |
| ----------------------------------------------------------- | |
| A tally over the 62 OpenSPP fixes the corpus was silent on said NINE of them | |
| added the name `AccessError` or `MissingError`, which read as the largest | |
| missing-rule cluster in that list. This checker was written from that count. | |
| It catches none of those nine. Reading the actual diffs afterwards -- which | |
| the count was standing in for, and should not have been -- the before-state | |
| is one handler, not two in the wrong order: | |
| try: | |
| service.submit(cr) | |
| except UserError as e: | |
| raise HTTPException(status_code=409, ...) | |
| AccessError is not mishandled there, it is unhandled: swallowed by the base | |
| class and reported as 409 Conflict. Nothing shadows anything, and "caught | |
| the base class without distinguishing a subclass" is not statically | |
| decidable as a defect, because `except UserError` is correct in most code. | |
| So the tally was a bad proxy for the diffs, which is the same mistake | |
| SCOPE_TOKENS makes when it counts a commit in scope for mentioning | |
| ValidationError. | |
| The construct this DOES catch is a real defect and the rule is kept for it, | |
| but with nothing behind it: zero findings on 229,600 lines of OpenSPP and no | |
| incident in any history measured here. It ships `advisory` on the second | |
| ground -- unproven rather than a preference -- which is exactly what that | |
| label is for. | |
| """ | |
| tree, _ = parse_python(code) | |
| if tree is None: | |
| return [] | |
| findings = [] | |
| for node in ast.walk(tree): | |
| if not isinstance(node, ast.Try): | |
| continue | |
| earlier = [] | |
| for handler in node.handlers: | |
| names = _handler_names(handler) | |
| for outer, outer_line in earlier: | |
| for inner in names: | |
| if inner != outer and _is_subclass_of(inner, outer): | |
| findings.append(( | |
| handler.lineno, | |
| "'except {0}' on line {1} already catches {2}, so " | |
| "this handler never runs -- {2} subclasses {0}".format( | |
| outer, outer_line, inner))) | |
| earlier.extend((name, handler.lineno) for name in names) | |
| return sorted(set(findings)) | |
| def depends_names_unread_fields(code): | |
| """odoo.depends-must-not-name-unread-fields | |
| The other direction of depends_covers_compute_reads, and a milder problem: | |
| a declared dependency the body never reads does not make the value stale, | |
| it makes the value recompute on writes that cannot change it. | |
| Harvested from OpenSPP's history rather than reasoned about. Two fix | |
| commits, both of them a leftover: | |
| @api.depends("partner_id", "pathway_id", "assessment_date") | |
| def _compute_name(self): # reads partner_id and pathway_id only | |
| @api.depends("topic_ids") | |
| def _compute_session_count(self): # counts rows on another model | |
| The second is why this is worth reporting at all. The decorator is the only | |
| statement anywhere of what a stored value depends on, and a reader trusts | |
| it. Left naming a field an older body used to read, it hides that this | |
| compute has no expressible dependency at all -- it queries a different | |
| model -- and the staleness that follows looks accounted for. | |
| Deliberately permissive, because a wrong entry costs a recompute and a | |
| false finding costs trust: | |
| A string literal anywhere in the body counts as a read. filtered("x"), | |
| mapped("a.b") and sorted(key=...) name fields without an attribute access | |
| ever appearing. | |
| A declared path is covered by any read that shares a prefix with it in | |
| either direction, so declaring line_ids while reading line_ids.price is | |
| fine and so is the reverse. | |
| A body that calls a method on self, or hands a record to any call, is | |
| left alone entirely. The read is then inside the callee, this pass cannot | |
| follow it, and the honest answer is not to guess. Measured on OpenSPP, | |
| this is the largest false-positive class: a compute that declares five | |
| fields and passes the record to a resolver that reads all five. | |
| A body that reads a field the decorator does not declare is left alone | |
| too. The two are already out of step, and that disagreement belongs to | |
| depends_covers_compute_reads -- reporting it twice from opposite | |
| directions is two findings about one confusion. It is also how a compute | |
| reading another stored compute looks from here, and those declare the | |
| upstream source on purpose. "Reads" here are the ones that rule makes, | |
| fields named in a string (_named_reads) among them, so the two agree | |
| about what the body reads. | |
| """ | |
| tree, _ = parse_python(code) | |
| if tree is None: | |
| return [] | |
| findings = [] | |
| for fn in _functions(tree): | |
| declared = _decorator_args(fn, "api", "depends") | |
| if not declared or declared is UNREADABLE: | |
| continue | |
| if _calls_a_method_on_self(fn) or _hands_a_record_to_a_call(fn) \ | |
| or _queries_another_model(fn) or _calls_super(fn): | |
| # Not exempt: a body that only assigns constants. On core four of | |
| # them were a stub for overrides or a deliberate reset, and on | |
| # OpenSPP and OCA three were leftovers classified true by hand -- | |
| # "Legacy method", a target model that is always res.partner. The | |
| # file cannot tell intent, so that shape stays a finding. | |
| continue | |
| aliases = _aliases(fn) | |
| read = set(".".join(path) for path, _ in | |
| _field_reads(fn, aliases) + _named_reads(fn, aliases)) | |
| produced = set(_assigned_attrs(fn)) | |
| if _reads_outside_the_declarations(read - produced, declared): | |
| continue | |
| read |= produced | |
| read |= _string_literals(fn) | |
| unread = [entry for entry in declared | |
| if not _covered_by_a_read(entry, read)] | |
| if unread: | |
| # One finding per method, not per entry. Three leftovers on one | |
| # decorator is one thing to fix, and three lines all pointing at | |
| # the same def read as three problems. | |
| findings.append(( | |
| fn.lineno, | |
| "{0}() declares {1} in @api.depends and never reads {2} -- the " | |
| "value recomputes on writes that cannot change it, and the " | |
| "decorator stops describing the method".format( | |
| fn.name, ", ".join(repr(str(e)) for e in unread), | |
| "them" if len(unread) > 1 else "it"))) | |
| return sorted(set(findings)) | |
| def _calls_super(fn): | |
| """Does the body call super()? | |
| Odoo merges @api.depends across the MRO: Field.get_depends walks every | |
| override of the compute and extends one list with each one's _depends | |
| (odoo/fields.py, 18.0 at 3c3e3b3). So an override's decorator ADDS | |
| triggers for the chain, and the fields it names are read by the parent's | |
| body, reached through super(). Core writes it as a comment: "EXTENDS 'web' | |
| to add dependencies". Its own body reading nothing is the point. Three of | |
| the fifteen sampled on odoo core were this, and all 40 findings the | |
| exemption removed from core were an override that delegates to super(). | |
| """ | |
| for node in _walk_body(fn): | |
| if (isinstance(node, ast.Call) and isinstance(node.func, ast.Name) | |
| and node.func.id == "super"): | |
| return True | |
| return False | |
| def _reads_outside_the_declarations(read, declared): | |
| """Does the body read a field no declared entry covers? | |
| Prefix-wise in both directions, so reading partner_id.name against a | |
| declared partner_id is inside the declarations and reading | |
| intervention_count against a declared intervention_ids is outside them. | |
| """ | |
| for path in read: | |
| if not any(path == entry or path.startswith(entry + ".") | |
| or entry.startswith(path + ".") for entry in declared): | |
| return True | |
| return False | |
| def _hands_a_record_to_a_call(fn): | |
| """Does any call in the body receive a record as an argument? | |
| resolve(scope) can read every field on scope, so a dependency declared for | |
| its benefit is correct and invisible here. A domain built from rec.id is | |
| not this: an id is an integer, and no callee reads a field through one. | |
| """ | |
| aliases = _aliases(fn) | |
| for node in _walk_body(fn): | |
| if not isinstance(node, ast.Call): | |
| continue | |
| for argument in list(node.args) + [kw.value for kw in node.keywords]: | |
| if _is_record_valued(argument, aliases): | |
| return True | |
| return False | |
| QUERY_METHODS = frozenset(["search", "search_count", "search_read", "read_group", | |
| "_read_group", "execute", "_search"]) | |
| def _queries_another_model(fn): | |
| """Does the body count or fetch rows from somewhere other than self? | |
| The largest class of false finding here by a wide margin: 17 of the 31 this | |
| reported on OpenSPP. A compute whose value comes from rows in ANOTHER model | |
| has no dependency @api.depends can express -- the ORM cannot watch a table | |
| this record has no field for -- so whatever the author declared is their | |
| best available approximation, usually the inverse One2many, and it often | |
| works. Telling them to delete it removes the only recompute trigger the | |
| value has. | |
| The cost of the exclusion is real and stated rather than hidden: one of the | |
| two harvested incidents this checker was written from is exactly this shape | |
| -- @api.depends("topic_ids") on a compute that counted rows on another | |
| model -- and is no longer reported. Reading it, the author's fix was to | |
| decide there was no correct dependency at all, which is a judgement this | |
| pass has no way to make for them. | |
| """ | |
| for node in _walk_body(fn): | |
| if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) \ | |
| and node.func.attr in QUERY_METHODS: | |
| return True | |
| return False | |
| def _is_record_valued(node, aliases): | |
| """Is this argument a record, or a field of one, rather than a plain value? | |
| Collection literals are looked inside, because a domain is a list of tuples | |
| and a vals dict is a dict of them. Attribute chains are NOT walked into: | |
| the Name at the base of rec.id is not itself an argument, and counting it | |
| made a search domain look like a record being handed over. | |
| """ | |
| if isinstance(node, ast.Name): | |
| return node.id in aliases | |
| if isinstance(node, ast.Attribute): | |
| path = _attr_path(node, aliases) | |
| return bool(path) and path[0] not in ORM_ATTRS \ | |
| and not path[0].startswith("_") | |
| if isinstance(node, (ast.List, ast.Tuple, ast.Set)): | |
| return any(_is_record_valued(item, aliases) for item in node.elts) | |
| if isinstance(node, ast.Dict): | |
| return any(_is_record_valued(item, aliases) | |
| for item in list(node.keys) + list(node.values) if item) | |
| return False | |
| def _covered_by_a_read(entry, read): | |
| """A declared path is covered by any read sharing a prefix with it.""" | |
| for path in read: | |
| if path == entry or path.startswith(entry + ".") \ | |
| or entry.startswith(path + "."): | |
| return True | |
| return False | |
| def _walk_body(fn): | |
| """Every node in the function's BODY, decorators excluded. | |
| ast.walk(fn) includes the decorator list, and @api.depends("x") holds the | |
| string "x". Counted as evidence the body mentions x, every declared entry | |
| proves itself and the checker silently finds nothing -- which is exactly | |
| what it did on the incident it was written from. | |
| """ | |
| for statement in fn.body: | |
| for node in ast.walk(statement): | |
| yield node | |
| def _string_literals(fn): | |
| """Every string in the body, plus each dotted path inside one. | |
| filtered("is_late") and mapped("line_ids.price") name fields without an | |
| attribute access ever appearing, and a declared dependency on either is | |
| correct. Splitting the dotted ones lets a declared prefix match too. | |
| """ | |
| out = set() | |
| for node in _walk_body(fn): | |
| if isinstance(node, ast.Constant) and isinstance(node.value, str): | |
| text = node.value.strip() | |
| if text: | |
| out.add(text) | |
| out.update(re.findall(r"[A-Za-z_][A-Za-z0-9_.]*", text)) | |
| return out | |
| def _calls_a_method_on_self(fn): | |
| """Does this body call a method ON the record -- rec._helper(), self._go()? | |
| The read is then inside the helper, where this pass cannot follow it, and a | |
| dependency declared for the helper's benefit is correct. Reporting it would | |
| be telling the author to delete a line that is doing its job. | |
| Length one, deliberately. rec._helper() resolves to the path ('_helper',) | |
| and can read anything; rec.line_ids.filtered(...) resolves to ('line_ids', | |
| 'filtered') and is an ORM call on one field, which reaches no other. Taking | |
| both lost the incident this checker was written from, where the body's only | |
| call was .replace() on a string. | |
| """ | |
| aliases = _aliases(fn) | |
| for node in _walk_body(fn): | |
| if not isinstance(node, ast.Call): | |
| continue | |
| func = node.func | |
| if not isinstance(func, ast.Attribute): | |
| continue | |
| path = _attr_path(func, aliases) | |
| if path is not None and len(path) == 1: | |
| return True | |
| return False | |
| # ------------------------------------------------ evidence for the triggers | |
| # | |
| # Two rules in the corpus are conditional on something no single file contains. | |
| # `ondelete` only bites if the parent is ever actually deleted; `check_company` | |
| # only bites if the codebase runs multi-company at all. Both conditions are | |
| # answerable from the tree, and these are the per-file halves of that answer -- | |
| # context.tree_facts aggregates them and the checkers read the aggregate. | |
| # | |
| # Everything here returns EVIDENCE FOUND, never evidence absent. One file not | |
| # deleting a model says nothing; the whole tree not deleting it is a claim, and | |
| # the only place entitled to make it is the thing that saw every file. | |
| # How a codebase shows that more than one company is really in play. A plain | |
| # `company_id` is not on this list on purpose: nearly every Odoo model declares | |
| # one, in single- and multi-company deployments alike, so it separates nothing. | |
| # These four are the machinery you only write when several companies exist -- | |
| # a user belonging to more than one, code reading the active set, or an author | |
| # who has already met the rule. | |
| MULTI_COMPANY_MARKERS = ("company_ids", "allowed_company_ids", | |
| "env.companies", "check_company") | |
| def models_declared(code): | |
| """Model names this file DEFINES, from `_name`. Not `_inherit`. | |
| The distinction decides what a tree is entitled to say about deletion. | |
| A model defined here is one whose code and access rules this tree owns, so | |
| "nothing in the tree deletes it" is a claim about the whole of it. A model | |
| merely extended with `_inherit` is somebody else's -- core Odoo's, usually -- | |
| and the delete could be anywhere in a codebase this walk never saw. | |
| """ | |
| tree, _ = parse_python(code) | |
| if tree is None: | |
| return set() | |
| names = set() | |
| for cls in [n for n in ast.walk(tree) if isinstance(n, ast.ClassDef)]: | |
| for node in cls.body: | |
| if not isinstance(node, ast.Assign) or len(node.targets) != 1: | |
| continue | |
| target = node.targets[0] | |
| if not (isinstance(target, ast.Name) and target.id == "_name"): | |
| continue | |
| if (isinstance(node.value, ast.Constant) | |
| and isinstance(node.value.value, str)): | |
| names.add(node.value.value) | |
| return names | |
| # Field types whose first positional argument names the comodel. For every other | |
| # type the first positional argument is the label. | |
| RELATIONAL_FIELDS = frozenset(["Many2one", "One2many", "Many2many"]) | |
| def orm_declarations(code): | |
| """What a file declares about its models, for the tree facts. | |
| One record per class that names a model: the model, the models it inherits | |
| (`_inherit` beyond itself, which put their methods in its MRO), the fields | |
| it declares and the literal @api.depends of each method that has one. | |
| Read the way Odoo reads them (odoo/models.py, 18.0 at 3c3e3b3): `_name`, | |
| or failing that a single `_inherit`; a class inheriting several models | |
| without a `_name` is named after itself there, and is left out here. | |
| A field is {"compute", "related", "store", "inverse", "comodel"}. `store` | |
| is True, False, None when the declaration does not say, or "?" when it | |
| says it with something other than a literal. A method's depends are the | |
| list, or None when @api.depends is there and cannot be read. `named` says | |
| the class gives the model its `_name`, rather than extending it. | |
| """ | |
| # ast.parse alone: the compile step parse_python adds reports errors this | |
| # never reads, and took about a quarter of the time (ARCHITECTURE 3.84). | |
| try: | |
| tree = ast.parse(code) | |
| except SyntaxError: | |
| return [] | |
| records = [] | |
| for cls in [n for n in ast.walk(tree) if isinstance(n, ast.ClassDef)]: | |
| model, named, inherit = _class_model(cls) | |
| if model is None: | |
| continue | |
| fields, methods = {}, {} | |
| for field, call in _class_field_calls(cls).items(): | |
| keywords = {k.arg: k.value for k in call.keywords if k.arg} | |
| store = keywords.get("store") | |
| compute = keywords.get("compute") | |
| comodel = None | |
| if call.func.attr in RELATIONAL_FIELDS: | |
| comodel = (_literal_string(call.args[0]) if call.args | |
| else _literal_string(keywords.get("comodel_name"))) | |
| fields[field] = { | |
| "compute": (compute.value if isinstance(compute, ast.Constant) | |
| and isinstance(compute.value, str) | |
| else compute.id if isinstance(compute, ast.Name) else None), | |
| "related": _literal_string(keywords.get("related")), | |
| "store": (None if store is None else store.value | |
| if isinstance(store, ast.Constant) else "?"), | |
| "inverse": "inverse" in keywords, | |
| "comodel": comodel, | |
| } | |
| for fn in cls.body: | |
| if not isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef)): | |
| continue | |
| depends = _decorator_args(fn, "api", "depends") | |
| if depends is not None: | |
| methods[fn.name] = None if depends is UNREADABLE else list(depends) | |
| records.append({"model": model, "named": named, | |
| "parents": [p for p in inherit if p != model], | |
| "fields": fields, "methods": methods}) | |
| return records | |
| def _literal_string(node): | |
| return node.value if isinstance(node, ast.Constant) and isinstance(node.value, str) \ | |
| else None | |
| def _literal_strings(node): | |
| """['a', 'b'] for "a" or ["a", "b"]; None when any part is not a literal.""" | |
| single = _literal_string(node) | |
| if single is not None: | |
| return [single] | |
| if isinstance(node, (ast.List, ast.Tuple)): | |
| values = [_literal_string(element) for element in node.elts] | |
| return values if all(v is not None for v in values) else None | |
| return None | |
| def models_unlinked(code): | |
| """Model names this file gives evidence of DELETING. | |
| Three shapes, which between them are how a delete is spelled in practice: | |
| self.env['x.y'].search(...).unlink() the model is named in the call | |
| self.unlink() inside a class, so the class's own | |
| models are what gets deleted | |
| def unlink(self) an override. Nobody writes one for a | |
| model nothing deletes, and it is the | |
| single clearest statement that | |
| deletion is a path this code expects. | |
| A local name assigned from `self.env['x.y']` and unlinked later resolves too, | |
| because splitting the search and the unlink across two lines is the common | |
| spelling and a checker that missed it would call the model undeleted. | |
| """ | |
| tree, _ = parse_python(code) | |
| if tree is None: | |
| return set() | |
| owners = {} # class node -> the models it speaks for | |
| for cls in [n for n in ast.walk(tree) if isinstance(n, ast.ClassDef)]: | |
| owners[cls] = _model_names(cls) | |
| found = set() | |
| # An override is a statement about the model the class declares. | |
| for cls, names in owners.items(): | |
| for node in cls.body: | |
| if (isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) | |
| and node.name == "unlink"): | |
| found |= names | |
| # Local names that hold a recordset of a known model. | |
| bound = {} | |
| for node in ast.walk(tree): | |
| if (isinstance(node, ast.Assign) and len(node.targets) == 1 | |
| and isinstance(node.targets[0], ast.Name)): | |
| model = _env_model_in(node.value) | |
| if model: | |
| bound[node.targets[0].id] = model | |
| for node in ast.walk(tree): | |
| if not (isinstance(node, ast.Call) | |
| and isinstance(node.func, ast.Attribute) | |
| and node.func.attr == "unlink"): | |
| continue | |
| receiver = node.func.value | |
| model = _env_model_in(receiver) | |
| if model: | |
| found.add(model) | |
| continue | |
| if isinstance(receiver, ast.Name) and receiver.id in bound: | |
| found.add(bound[receiver.id]) | |
| continue | |
| # `self.unlink()` and `record.unlink()` inside a model body: the models | |
| # the enclosing class declares are the ones being deleted. | |
| if isinstance(receiver, ast.Name) and receiver.id == "self": | |
| found |= _enclosing_models(tree, node, owners) | |
| return found | |
| def _env_model_in(node): | |
| """The model name in an `env['x.y']` subscript anywhere inside node, if the | |
| expression is rooted in one. `self.env['x'].search(...)` and | |
| `self.env['x'].browse(i)` both resolve; a bare `records` does not.""" | |
| if node is None: | |
| return None | |
| for inner in ast.walk(node): | |
| if not isinstance(inner, ast.Subscript): | |
| continue | |
| value = inner.value | |
| if not (isinstance(value, ast.Attribute) and value.attr == "env"): | |
| continue | |
| key = inner.slice | |
| if isinstance(key, ast.Constant) and isinstance(key.value, str): | |
| return key.value | |
| return None | |
| def _enclosing_models(tree, target, owners): | |
| """The models declared by the class that lexically contains `target`.""" | |
| for cls, names in owners.items(): | |
| for node in ast.walk(cls): | |
| if node is target: | |
| return names | |
| return set() | |
| def declares_multi_company(code): | |
| """True if this file shows the codebase really runs more than one company. | |
| Textual on purpose. The markers are attribute names and a keyword argument | |
| rather than a shape, so an AST pass would spell the same search at more | |
| length, and the question is only ever "does this appear anywhere". | |
| """ | |
| view = strip_comments(code, "python") | |
| return any(marker in view for marker in MULTI_COMPANY_MARKERS) | |
| # ------------------------------------------- where a value came from | |
| # | |
| # Four rules were classified false for the same reason: each matched a | |
| # variable's NAME and called that evidence about its value. `total == 0` is a | |
| # float comparison in one function and an exact integer test in the next; | |
| # `filename` says nothing about whether a caller chose it. Telling those apart | |
| # needs the assignment, which is what this section reads. | |
| # | |
| # It is a local trace and nothing more. One file, one function, no imports | |
| # followed and no types inferred -- which is enough for the shapes that were | |
| # actually misread, and the limit is stated here rather than discovered later. | |
| def _function_of(tree): | |
| """Maps every node to the function that encloses it, innermost winning. | |
| Lambdas count. `filtered(lambda x: x.qty == 0)` is where one of the | |
| misreadings lived, and a lambda binds its arguments exactly as a def does. | |
| """ | |
| owner = {} | |
| scopes = [n for n in ast.walk(tree) | |
| if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef, ast.Lambda))] | |
| # Outermost first, so an inner scope overwrites the outer one's claim. | |
| scopes.sort(key=lambda n: (getattr(n, "lineno", 0), | |
| getattr(n, "col_offset", 0))) | |
| for fn in scopes: | |
| for node in ast.walk(fn): | |
| owner[id(node)] = fn | |
| return owner | |
| def _final_name(node): | |
| """The last identifier in a Name or Attribute, or None for anything else.""" | |
| if isinstance(node, ast.Name): | |
| return node.id | |
| if isinstance(node, ast.Attribute): | |
| return node.attr | |
| return None | |
| def _chain_root(node): | |
| """The leftmost Name in a chain of attributes, calls and subscripts.""" | |
| while True: | |
| if isinstance(node, ast.Attribute): | |
| node = node.value | |
| elif isinstance(node, ast.Subscript): | |
| node = node.value | |
| elif isinstance(node, ast.Call): | |
| node = node.func | |
| else: | |
| break | |
| return node.id if isinstance(node, ast.Name) else None | |
| def _targets(node): | |
| """Every Name a statement binds, whatever the binding form.""" | |
| found = [] | |
| for sub in ast.walk(node): | |
| if isinstance(sub, ast.Name) and isinstance(sub.ctx, (ast.Store, ast.Del)): | |
| found.append(sub.id) | |
| return found | |
| def _bound_in(fn): | |
| """Names this scope binds: parameters, assignments, loops, withs, imports.""" | |
| if fn is None: | |
| return set() | |
| names = set() | |
| args = getattr(fn, "args", None) | |
| if args is not None: | |
| for group in (args.args, args.posonlyargs, args.kwonlyargs): | |
| names.update(a.arg for a in group or []) | |
| for extra in (args.vararg, args.kwarg): | |
| if extra is not None: | |
| names.add(extra.arg) | |
| for node in ast.walk(fn): | |
| if isinstance(node, (ast.Assign, ast.AnnAssign, ast.AugAssign, ast.For, | |
| ast.AsyncFor, ast.comprehension, ast.NamedExpr)): | |
| names.update(_targets(node)) | |
| elif isinstance(node, (ast.With, ast.AsyncWith)): | |
| for item in node.items: | |
| if item.optional_vars is not None: | |
| names.update(_targets(item.optional_vars)) | |
| elif isinstance(node, ast.ExceptHandler) and node.name: | |
| names.add(node.name) | |
| elif isinstance(node, (ast.Import, ast.ImportFrom)): | |
| names.update((a.asname or a.name).split(".")[0] for a in node.names) | |
| elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): | |
| if node is not fn: | |
| names.add(node.name) | |
| return names | |
| def _bound_at_module(tree): | |
| """Names the file itself binds at any level outside a function body.""" | |
| names = set() | |
| for node in ast.walk(tree): | |
| if isinstance(node, (ast.Assign, ast.AnnAssign, ast.AugAssign)): | |
| names.update(_targets(node)) | |
| elif isinstance(node, (ast.Import, ast.ImportFrom)): | |
| names.update((a.asname or a.name).split(".")[0] for a in node.names) | |
| elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): | |
| names.add(node.name) | |
| return names | |
| # Reads that are external by their own name, whatever the surrounding file | |
| # does with them. `request` is werkzeug's, reached through odoo.http; the | |
| # context is whatever the caller put there; both parsers turn somebody else's | |
| # bytes into objects. | |
| EXTERNAL_ROOTS = frozenset(["request"]) | |
| EXTERNAL_CALLS = frozenset(["loads", "literal_eval", "parse_qs", "parse_qsl"]) | |
| CONTEXT_READS = ("_context", "context") | |
| def _is_external(node, fn, tree): | |
| """True when this expression's value came from outside the file. | |
| Two ways to qualify. Either the expression names a source that is external | |
| by construction -- `request`, a parsed payload, the context a caller | |
| handed in -- or its root name is one this file never binds, which is the | |
| same answer arrived at by elimination: a name that is not a parameter, not | |
| assigned, not a loop or with target, not imported and not a builtin is a | |
| value from somewhere this file cannot see. | |
| The elimination half is what makes the rules' own 'wrong' samples fire, | |
| because a sample is a fragment and its inputs are free names by | |
| definition. It is also what keeps them quiet on the four findings that | |
| were classified false: `file_config`, `filename`, `full_path` and | |
| `raw_value` are all bound a few lines above the join. | |
| """ | |
| for sub in ast.walk(node): | |
| if isinstance(sub, ast.Attribute) and sub.attr in CONTEXT_READS: | |
| return True | |
| if isinstance(sub, ast.Call): | |
| attr = _final_name(sub.func) | |
| if attr in EXTERNAL_CALLS: | |
| return True | |
| root = _chain_root(node) | |
| if root is None: | |
| return False | |
| if root in EXTERNAL_ROOTS: | |
| return True | |
| if root in ("self", "cls"): | |
| return False | |
| if root in dir(builtins): | |
| return False | |
| return root not in _bound_in(fn) and root not in _bound_at_module(tree) | |
| # Calls whose result is a whole number, so `== 0` on it is exact and the | |
| # float rule has nothing to say. search_count and len are counts by | |
| # definition; `.count()` is one on a recordset, a string and a list alike. | |
| INT_VALUED_CALLS = frozenset(["search_count", "count", "len", "bit_length", | |
| "index", "ord"]) | |
| _COUNT_KEY = re.compile(r"(?:^__count$|_count$)") | |
| def _integer_fields(tree): | |
| """Field names the file declares as fields.Integer.""" | |
| names = set() | |
| for node in ast.walk(tree): | |
| if not isinstance(node, ast.Assign) or not isinstance(node.value, ast.Call): | |
| continue | |
| func = node.value.func | |
| if isinstance(func, ast.Attribute) and func.attr == "Integer": | |
| names.update(_targets(node)) | |
| return names | |
| def _counts_rows(fn): | |
| """True when this scope runs a COUNT query, so a fetched cell is an int.""" | |
| if fn is None: | |
| return False | |
| for node in ast.walk(fn): | |
| if not (isinstance(node, ast.Call) and _final_name(node.func) == "execute"): | |
| continue | |
| for arg in node.args[:1]: | |
| try: | |
| text = ast.unparse(arg) | |
| except Exception: # pragma: no cover | |
| continue | |
| if "count(" in text.lower(): | |
| return True | |
| return False | |
| def _assigned_values(fn, name): | |
| """Values assigned to `name` in this scope, or None if it is bound some | |
| way the trace cannot read -- a loop target, a with, an unpacking.""" | |
| if fn is None: | |
| return None | |
| values = [] | |
| for node in ast.walk(fn): | |
| if isinstance(node, ast.Assign): | |
| for target in node.targets: | |
| if isinstance(target, ast.Name) and target.id == name: | |
| values.append(node.value) | |
| elif name in _targets(target): | |
| return None # unpacked: no single value | |
| elif isinstance(node, ast.AnnAssign): | |
| if isinstance(node.target, ast.Name) and node.target.id == name \ | |
| and node.value is not None: | |
| values.append(node.value) | |
| elif isinstance(node, (ast.For, ast.AsyncFor, ast.comprehension)): | |
| if name in _targets(node.target): | |
| return None | |
| elif isinstance(node, (ast.With, ast.AsyncWith)): | |
| for item in node.items: | |
| if item.optional_vars is not None \ | |
| and name in _targets(item.optional_vars): | |
| return None | |
| elif isinstance(node, ast.AugAssign): | |
| if isinstance(node.target, ast.Name) and node.target.id == name: | |
| values.append(node.value) | |
| return values or None | |
| def _int_valued(node, fn, tree, depth=0): | |
| """True when this expression is certainly a whole number. | |
| Certainly, not probably: every answer here SUPPRESSES a finding, so the | |
| unreadable cases must come back False and leave the finding standing. A | |
| name assigned twice is only an integer if both assignments are. | |
| """ | |
| if depth > 4: # pragma: no cover | |
| return False | |
| if isinstance(node, ast.Constant): | |
| return isinstance(node.value, int) and not isinstance(node.value, bool) | |
| if isinstance(node, ast.Call): | |
| if _final_name(node.func) in INT_VALUED_CALLS: | |
| return True | |
| return False | |
| if isinstance(node, ast.Subscript): | |
| # read_group hands back rows keyed '<field>_count', and the key is | |
| # written as an f-string as often as a literal. | |
| key = node.slice | |
| if isinstance(key, ast.Constant) and isinstance(key.value, str): | |
| if _COUNT_KEY.search(key.value): | |
| return True | |
| if isinstance(key, ast.JoinedStr) and key.values: | |
| tail = key.values[-1] | |
| if isinstance(tail, ast.Constant) and isinstance(tail.value, str) \ | |
| and _COUNT_KEY.search(tail.value): | |
| return True | |
| # cr.fetchone()[0] of a SELECT COUNT(*) is the third spelling of a | |
| # count, and the query is the only place that says so. | |
| inner = node.value | |
| if isinstance(inner, ast.Call) and _final_name(inner.func) in ( | |
| "fetchone", "fetchall") and _counts_rows(fn): | |
| return True | |
| return False | |
| if isinstance(node, ast.BinOp) and isinstance(node.op, (ast.Add, ast.Sub, | |
| ast.Mult, ast.Mod, | |
| ast.FloorDiv)): | |
| return (_int_valued(node.left, fn, tree, depth + 1) | |
| and _int_valued(node.right, fn, tree, depth + 1)) | |
| if isinstance(node, ast.Attribute): | |
| return node.attr in _integer_fields(tree) | |
| if isinstance(node, ast.Name): | |
| if node.id in _integer_fields(tree): | |
| return True | |
| values = _assigned_values(fn, node.id) | |
| if not values: | |
| return False | |
| return all(_int_valued(v, fn, tree, depth + 1) for v in values) | |
| return False | |
| # ------------------------------------------------ the rewritten checkers | |
| _MONEY_NAME = re.compile(r"price|amount|qty|quantity|total|balance", re.I) | |
| _RESOLVES_THE_PATH = frozenset(["commonpath", "commonprefix", "relpath", | |
| "is_relative_to", "realpath", "abspath"]) | |
| def float_equality_against_zero(code): | |
| """odoo.float-comparison-needs-float-utils | |
| `== 0` or `!= 0` on a value whose name says it is money or a quantity, | |
| where the assignment does not say it is a whole number. | |
| The regex this replaces asked only the first half. Classified by hand it | |
| was 2 true and 5 false, and every one of the five compared an INTEGER to | |
| zero, where `==` is exact and the rule has nothing to say: | |
| `search_count(...)`, a read_group `<field>_count`, `cr.fetchone()[0]` of a | |
| SELECT COUNT(*), and `x.qty` where the same file declares qty as | |
| fields.Integer. All five were named `total`, `group_total` or | |
| `total_population`, and so were both true positives -- `total = | |
| sum(sorted_amounts)` is the same spelling with a float behind it. The name | |
| could never have separated them. | |
| What it cannot see, and so still reports: a float that arrives as a | |
| parameter, and an integer from a helper in another file. Both leave the | |
| finding standing, which is the direction a suppressing trace has to err in. | |
| """ | |
| tree, _ = parse_python(code) | |
| if tree is None: | |
| return [] | |
| owner = _function_of(tree) | |
| findings, seen = [], set() | |
| for node in ast.walk(tree): | |
| if not isinstance(node, ast.Compare) or len(node.ops) != 1: | |
| continue | |
| if not isinstance(node.ops[0], (ast.Eq, ast.NotEq)): | |
| continue | |
| left, right = node.left, node.comparators[0] | |
| for value, other in ((left, right), (right, left)): | |
| if not (isinstance(other, ast.Constant) | |
| and isinstance(other.value, (int, float)) | |
| and not isinstance(other.value, bool) | |
| and other.value == 0): | |
| continue | |
| name = _final_name(value) | |
| if not name or not _MONEY_NAME.search(name): | |
| continue | |
| if _int_valued(value, owner.get(id(value)), tree): | |
| break | |
| if node.lineno in seen: | |
| break | |
| seen.add(node.lineno) | |
| findings.append(( | |
| node.lineno, | |
| "'{0}' is compared with == against zero -- binary floats do " | |
| "not land on it exactly, so use float_is_zero with the " | |
| "field's precision".format(name), | |
| None)) | |
| break | |
| return findings | |
| def path_join_takes_an_external_name(code): | |
| """python.path-join-does-not-contain-a-path | |
| os.path.join where one component came from outside the file, and the | |
| result is not resolved and compared before it is used. | |
| The regex this replaces matched component NAMES -- `filename`, `form`, | |
| `params` -- and was 0 true, 4 false. All four join a configured directory | |
| to a name the module itself produced: a timestamped backup filename, a | |
| literal from a list declared in the same function, a module path out of | |
| get_module_path. Not one was a fragment a caller could point out of the | |
| directory, which is the only thing the rule is about. | |
| So the question moved from what the component is called to where its value | |
| came from, and 'outside the file' is the answer _is_external gives. The | |
| rule's own wrong sample joins `request.args['name']` and still fires. | |
| This reports less than the regex did, on purpose: the regex's entire | |
| output on 294,000 lines was false, and a checker that cries wolf gets | |
| switched off. A traversal reached through a helper in another module is | |
| outside what a local trace can see and is not reported. | |
| """ | |
| tree, _ = parse_python(code) | |
| if tree is None: | |
| return [] | |
| owner = _function_of(tree) | |
| findings = [] | |
| for node in ast.walk(tree): | |
| if not (isinstance(node, ast.Call) and _final_name(node.func) == "join"): | |
| continue | |
| # os.path.join, not str.join and not pathlib's. | |
| parent = node.func.value if isinstance(node.func, ast.Attribute) else None | |
| if not (isinstance(parent, ast.Attribute) and parent.attr == "path"): | |
| continue | |
| fn = owner.get(id(node)) | |
| # The fix is to resolve and compare, and it is written beside the | |
| # join. Scoped to the enclosing function rather than the file: a | |
| # realpath somewhere else in a 900-line model says nothing about this | |
| # call. A join at module scope -- which is what a rule's own sample | |
| # is -- is scoped to the module for the same reason. | |
| scope = fn if fn is not None else tree | |
| if any(_final_name(c.func) in _RESOLVES_THE_PATH | |
| for c in ast.walk(scope) if isinstance(c, ast.Call)): | |
| continue | |
| for arg in node.args[1:]: | |
| if _is_external(arg, fn, tree): | |
| try: | |
| shown = ast.unparse(arg) | |
| except Exception: # pragma: no cover | |
| shown = "the component" | |
| findings.append(( | |
| node.lineno, | |
| "os.path.join is handed {0}, which came from outside this " | |
| "file -- an absolute component discards the base directory " | |
| "entirely and '..' walks out of it, so resolve both and " | |
| "compare before opening anything".format(shown), | |
| None)) | |
| break | |
| return findings | |
| # Modules whose presence in the imports makes a file an HTTP edge. Not | |
| # werkzeug: Odoo imports werkzeug.urls to build links in model files, and one | |
| # such file on core was the only thing a wider gate added. | |
| _HTTP_MODULES = ("fastapi", "pydantic", "odoo.http", "starlette") | |
| def _datetime_class_names(tree): | |
| """Names the datetime CLASS is bound to: `datetime`, plus any alias.""" | |
| names = {"datetime"} | |
| for node in ast.walk(tree): | |
| if isinstance(node, ast.ImportFrom) and node.module == "datetime": | |
| names.update(a.asname for a in node.names | |
| if a.name == "datetime" and a.asname) | |
| return names | |
| def _speaks_http(tree): | |
| for node in ast.walk(tree): | |
| if isinstance(node, ast.ImportFrom) and node.module: | |
| if node.module.startswith(_HTTP_MODULES): | |
| return True | |
| if node.module == "odoo" and any(a.name == "http" for a in node.names): | |
| return True | |
| elif isinstance(node, ast.Import): | |
| if any(a.name.startswith(_HTTP_MODULES) for a in node.names): | |
| return True | |
| return False | |
| class _NaiveNow(object): | |
| """Which expressions carry a naive now(), and as a time or as a date. | |
| The sources are the ones the regex matched -- datetime.now() and utcnow() | |
| with empty parentheses -- and nothing more. date.today() and | |
| fields.Date.today() are the same UTC date and were measured rather than | |
| added: 48 and 114 sites on OpenSPP, 31 and 217 on core. Adding them would | |
| turn a narrowing into a widening, and fields.Date.today() is the | |
| framework's own spelling of the value. | |
| """ | |
| def __init__(self, tree): | |
| self.names = _datetime_class_names(tree) | |
| self.owner = _function_of(tree) | |
| self.at = None | |
| def _is_source(self, node): | |
| if not (isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute)): | |
| return False | |
| if node.args or node.keywords: | |
| return False | |
| if node.func.attr == "utcnow": | |
| return True | |
| return node.func.attr == "now" and _final_name(node.func.value) in self.names | |
| def carries(self, node, depth=0): | |
| """None, "datetime" or "date"; the source's line is left in self.at.""" | |
| if node is None or depth > 5: | |
| return None | |
| if self._is_source(node): | |
| self.at = node.lineno | |
| return "datetime" | |
| if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute): | |
| inner = self.carries(node.func.value, depth + 1) | |
| if node.func.attr == "date" and not node.args and inner: | |
| return "date" | |
| if node.func.attr == "replace" and inner: | |
| # .replace(tzinfo=...) is the fix; any other replace keeps it naive. | |
| return None if any(k.arg == "tzinfo" for k in node.keywords) else inner | |
| if node.func.attr == "get" and len(node.args) == 2: | |
| # vals.get("date", datetime.now()): the default is the value | |
| # whenever the caller did not supply one. | |
| return self.carries(node.args[1], depth + 1) | |
| return None | |
| if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Add): | |
| return self.carries(node.left, depth + 1) or self.carries(node.right, depth + 1) | |
| if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Sub): | |
| # now - timedelta is a time; now - then is a duration. | |
| left = self.carries(node.left, depth + 1) | |
| return None if self.carries(node.right, depth + 1) else left | |
| if isinstance(node, (ast.BoolOp, ast.IfExp)): | |
| parts = node.values if isinstance(node, ast.BoolOp) else [node.body, node.orelse] | |
| for part in parts: | |
| found = self.carries(part, depth + 1) | |
| if found: | |
| return found | |
| return None | |
| if isinstance(node, ast.Name): | |
| for value in _assigned_values(self.owner.get(id(node)), node.id) or (): | |
| found = self.carries(value, depth + 1) | |
| if found: | |
| return found | |
| return None | |
| def serialised(self, node): | |
| """`x.isoformat()` or `str(x)` of a naive value: text with no offset.""" | |
| if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) \ | |
| and node.func.attr == "isoformat" and not node.args: | |
| return self.carries(node.func.value) | |
| if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) \ | |
| and node.func.id == "str" and node.args: | |
| return self.carries(node.args[0]) | |
| return None | |
| def _offset_written_by_hand(parent_of, node): | |
| """`f"{now.isoformat()}Z"` or `now.isoformat() + "Z"`: labelled UTC.""" | |
| up = parent_of.get(id(node)) | |
| while isinstance(up, ast.FormattedValue): | |
| up = parent_of.get(id(up)) | |
| tails = [] | |
| if isinstance(up, ast.JoinedStr): | |
| tails = [v.value for v in up.values if isinstance(v, ast.Constant)] | |
| elif isinstance(up, ast.BinOp) and isinstance(up.op, ast.Add) \ | |
| and isinstance(up.right, ast.Constant): | |
| tails = [up.right.value] | |
| return any(isinstance(t, str) and t.startswith(("Z", "+00")) for t in tails) | |
| def _names_a_date(key): | |
| """A vals key or domain field spelled as a date: date, date_*, *_date.""" | |
| if not (isinstance(key, ast.Constant) and isinstance(key.value, str)): | |
| return False | |
| k = key.value | |
| return "datetime" not in k and (k == "date" or k.startswith("date_") | |
| or k.endswith("_date")) | |
| def naive_time_leaves_or_dates_a_day(code): | |
| """python.naive-and-aware-datetimes-do-not-compare | |
| The two shapes that survived reading, and only those. | |
| The regex this replaces fired on every datetime.now() and utcnow(). Its | |
| premise was that a naive local time reaching the ORM is wrong by the | |
| server's offset, and inside Odoo there is no offset: `import odoo` pins | |
| TZ=UTC, and fields.Datetime.now() is datetime.now() itself (ARCHITECTURE | |
| 3.56). Re-read on that basis it was 3 true of 33 on OpenSPP and OCA and 0 | |
| of 15 sampled on core. What stands does not depend on the server clock: | |
| A. A naive timestamp LEAVING the system with no offset: its isoformat() | |
| anywhere, or the value itself as a keyword of a class being built (a | |
| pydantic schema, a response object) or in a dict the function | |
| returns, in a file that imports fastapi, pydantic, starlette or | |
| odoo.http. The consumer cannot tell the zone. Two consent API | |
| responses on OpenSPP, both commits recall credits the rule with, and | |
| the xades:SigningTime of two Spanish e-invoicing modules on core. | |
| B. The server's UTC DATE used as a calendar day: a .date() of it into a | |
| vals key or domain field spelled as a date, a Date field's default, | |
| or an attribute named for a date; and a naive now() itself into a key | |
| named exactly `date`, which Odoo truncates to UTC's day. OCA's | |
| purchase_unreconciled dates a write-off that way, so for a company | |
| ahead of UTC the early hours land on the day before. | |
| What it leaves alone on purpose: a comparison (the fiscal-year cron was a | |
| day ahead by design), a filename or label (display only), a JWT payload | |
| (PyJWT reads naive as UTC), an ISO string with a Z written after it, and a | |
| value that only meets other naive UTC values inside the ORM. A key ending | |
| in _date can be a Datetime field (stock's scheduled_date), which is why a | |
| naive now() into one is not reported and its .date() is -- and why | |
| l10n_in's demo invoices, `'invoice_date': datetime.now()` into a Date | |
| field, are missed. The isoformat() clause costs three findings OpenSPP's | |
| classification reads false: a generated-at line in two report scripts and | |
| a computed_at in a result dict. | |
| """ | |
| tree, _ = parse_python(code) | |
| if tree is None: | |
| return [] | |
| now = _NaiveNow(tree) | |
| parent_of = {} | |
| for node in ast.walk(tree): | |
| for child in ast.iter_child_nodes(node): | |
| parent_of[id(child)] = node | |
| returned = set() | |
| for fn in _functions(tree): | |
| for node in ast.walk(fn): | |
| if isinstance(node, ast.Return) and node.value is not None: | |
| returned.add(id(node.value)) | |
| if isinstance(node.value, ast.Name): | |
| returned.update(id(v) for v in _assigned_values(fn, node.value.id) or ()) | |
| findings, seen = [], set() | |
| def report(text): | |
| if now.at not in seen: | |
| seen.add(now.at) | |
| findings.append((now.at, text, None)) | |
| # An ISO string is text for somebody else to read, wherever it is built. | |
| # The HTTP gate alone lost two on core: the Spanish e-invoicing modules | |
| # write datetime.now().isoformat() into xades:SigningTime, an xsd:dateTime | |
| # that a reader with no offset takes as local time. | |
| for node in ast.walk(tree): | |
| if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) \ | |
| and node.func.attr == "isoformat" and now.serialised(node) \ | |
| and not _offset_written_by_hand(parent_of, node): | |
| report("isoformat() of a naive timestamp writes no offset, so whoever " | |
| "reads the text cannot tell which zone it is in -- write " | |
| "datetime.now(timezone.utc) at the edge") | |
| if _speaks_http(tree): | |
| for node in ast.walk(tree): | |
| if isinstance(node, ast.Call) and (_final_name(node.func) or "")[:1].isupper(): | |
| values = [k.value for k in node.keywords] | |
| where = "into {0}()".format(_final_name(node.func)) | |
| elif isinstance(node, ast.Dict) and id(node) in returned: | |
| values = [v for v in node.values if v is not None] | |
| where = "in a returned dict" | |
| else: | |
| continue | |
| for value in values: | |
| if (now.carries(value) or now.serialised(value)) \ | |
| and not _offset_written_by_hand(parent_of, value): | |
| report("a naive timestamp leaves {0} with no offset, so whoever " | |
| "reads it cannot tell which zone it is in -- write " | |
| "datetime.now(timezone.utc) at the edge".format(where)) | |
| for node in ast.walk(tree): | |
| if isinstance(node, ast.Dict): | |
| for key, value in zip(node.keys, node.values): | |
| if not _names_a_date(key): | |
| continue | |
| kind = now.carries(value) | |
| if kind == "date" or (kind == "datetime" and key.value == "date"): | |
| report("'{0}' is set to the server's UTC date -- for a company " | |
| "ahead of UTC that is yesterday in its early hours; " | |
| "fields.Date.context_today is the user's day".format(key.value)) | |
| elif isinstance(node, ast.Tuple) and len(node.elts) == 3 \ | |
| and _names_a_date(node.elts[0]): | |
| if now.carries(node.elts[2]) == "date": | |
| report("a domain on '{0}' compares against the server's UTC date " | |
| "rather than the user's -- fields.Date.context_today is " | |
| "the user's day".format(node.elts[0].value)) | |
| elif isinstance(node, ast.Call) and _final_name(node.func) == "Date": | |
| for kw in node.keywords: | |
| body = kw.value.body if isinstance(kw.value, ast.Lambda) else kw.value | |
| if kw.arg == "default" and now.carries(body): | |
| report("a Date field defaults to the server's UTC date -- " | |
| "fields.Date.context_today is the user's day") | |
| elif isinstance(node, ast.Assign) and len(node.targets) == 1 \ | |
| and isinstance(node.targets[0], ast.Attribute): | |
| attr = node.targets[0].attr | |
| if "datetime" not in attr and "date" in attr.split("_") \ | |
| and now.carries(node.value) == "date": | |
| report("'{0}' is set to the server's UTC date -- " | |
| "fields.Date.context_today is the user's day".format(attr)) | |
| return sorted(findings) | |
| def float_coercion_of_external_money(code): | |
| """python.decimal-for-money-not-float | |
| float() applied to a money value that arrived from outside the file. | |
| The regex this replaces matched the NAME either side of the coercion and | |
| was 0 true, 3 false, with a reason behind it rather than a sample size: | |
| Odoo's Monetary field IS a float -- the ORM defines it that way -- and the | |
| framework's answer to the precision problem is float_round and | |
| float_compare against the currency's rounding, not Decimal. Inside an Odoo | |
| module the rule fired on the idiom and named a fix that cannot be taken. | |
| Three findings do not settle a rule either way, so it is narrowed rather | |
| than culled, and the narrowing is the line the argument actually draws. | |
| `float(self.amount or 0.0)` is the ORM's own type and is not reported. | |
| `float(result)`, where result came back from a service in this file, is | |
| not reported. `float(payload['price'])`, where the string came from a | |
| caller and Decimal(payload['price']) IS available, still is -- which is | |
| the rule's own wrong sample and the one shape its remedy fits. | |
| """ | |
| tree, _ = parse_python(code) | |
| if tree is None: | |
| return [] | |
| owner = _function_of(tree) | |
| findings = [] | |
| for node in ast.walk(tree): | |
| if not (isinstance(node, ast.Call) and isinstance(node.func, ast.Name) | |
| and node.func.id == "float" and node.args): | |
| continue | |
| value = node.args[0] | |
| name = _final_name(value) or _chain_root(value) or "" | |
| target = _MONEY_NAME.search(name) | |
| if not target: | |
| # The coerced expression may say nothing, while what it is | |
| # assigned to does: `price = float(payload['p'])`. | |
| holder = owner.get(id(node)) | |
| for stmt in ast.walk(holder if holder is not None else tree): | |
| if isinstance(stmt, ast.Assign) and stmt.value is node: | |
| target = any(_MONEY_NAME.search(t) | |
| for t in _targets(stmt) if t) | |
| break | |
| if not target: | |
| continue | |
| if not _is_external(value, owner.get(id(node)), tree): | |
| continue | |
| findings.append(( | |
| node.lineno, | |
| "a money value from outside this file is routed through float() " | |
| "-- the binary float is already wrong before Decimal can be given " | |
| "it, so build Decimal from the original string", | |
| None)) | |
| return findings | |
| class _Every(object): | |
| """A constraint is present and which fields it covers cannot be read.""" | |
| EVERY_FIELD = _Every() | |
| def _sql_identifiers(node): | |
| """Names inside a constraint's SQL text, or EVERY_FIELD when the text is | |
| not a literal. Keywords come along (CHECK, UNIQUE, NOT) and match no field.""" | |
| if isinstance(node, ast.Constant) and isinstance(node.value, str): | |
| return set(re.findall(r"[A-Za-z_][A-Za-z0-9_]*", node.value)) | |
| return EVERY_FIELD | |
| def _constrained_fields(tree, legacy_dead): | |
| """Field names some constraint in this file covers, or EVERY_FIELD. | |
| Three spellings: @api.constrains (its plain names -- a dotted one registers | |
| nothing), models.Constraint (the names in its SQL) and _sql_constraints (the | |
| names in each entry's SQL, and nothing at all when `legacy_dead`, because | |
| from 19 the registry never creates it). A constraint that is present but | |
| cannot be read -- @api.constrains(*FIELDS), a _sql_constraints inherited and | |
| extended -- covers every field, which is the reading this had before it | |
| looked at fields at all. | |
| """ | |
| covered = set() | |
| def add(names): | |
| if names is EVERY_FIELD: | |
| return True | |
| covered.update(names) | |
| return False | |
| for fn in _functions(tree): | |
| declared = _decorator_args(fn, "api", "constrains") | |
| if declared is UNREADABLE: | |
| return EVERY_FIELD | |
| if declared: | |
| covered.update(d for d in declared if "." not in d) | |
| for node in ast.walk(tree): | |
| if isinstance(node, ast.Call): | |
| func = node.func | |
| name = func.attr if isinstance(func, ast.Attribute) else \ | |
| func.id if isinstance(func, ast.Name) else None | |
| if name == "Constraint": | |
| if add(_sql_identifiers(node.args[0]) if node.args else EVERY_FIELD): | |
| return EVERY_FIELD | |
| continue | |
| if isinstance(node, (ast.Assign, ast.AugAssign, ast.AnnAssign)): | |
| targets = node.targets if isinstance(node, ast.Assign) else [node.target] | |
| if not any(isinstance(t, ast.Name) and t.id == "_sql_constraints" | |
| for t in targets) or node.value is None or legacy_dead: | |
| continue | |
| if not isinstance(node.value, (ast.List, ast.Tuple)): | |
| return EVERY_FIELD | |
| for entry in node.value.elts: | |
| if not (isinstance(entry, (ast.List, ast.Tuple)) and len(entry.elts) > 1): | |
| return EVERY_FIELD | |
| if add(_sql_identifiers(entry.elts[1])): | |
| return EVERY_FIELD | |
| return covered | |
| def _clears(value): | |
| """False, None, '' or an empty collection: what a form resets a field to.""" | |
| if isinstance(value, ast.Constant): | |
| return value.value is False or value.value is None or value.value == "" | |
| if isinstance(value, ast.Dict): | |
| return not value.keys | |
| return isinstance(value, (ast.List, ast.Tuple, ast.Set)) and not value.elts | |
| def _refuses_in_branch(fn, warning, watched): | |
| """Does the branch that returns this warning also take the value back? | |
| A warning dict on its own informs: "this reference already exists", "the | |
| rate is far from the previous one", "please choose another package type". | |
| The user saves anyway, from the form as from an import, so nothing is | |
| enforced anywhere and the rule's harm -- a check only the form applies -- | |
| is not there. What refuses is the same branch writing back a field the | |
| onchange watches (the quantity clamped, the partner cleared, the removed | |
| document restored) or clearing any field on the record (a registrant that | |
| does not fit the new request type). "The same branch" is the innermost | |
| if, for or while around the warning, or the whole body when there is none. | |
| """ | |
| parents = {} | |
| for node in ast.walk(fn): | |
| for child in ast.iter_child_nodes(node): | |
| parents[id(child)] = node | |
| branch = warning | |
| while branch is not fn and not isinstance(branch, (ast.If, ast.For, ast.AsyncFor, | |
| ast.While)): | |
| branch = parents.get(id(branch), fn) | |
| roots = set() if watched is UNREADABLE else {w.split(".")[0] for w in watched} | |
| aliases = _aliases(fn) | |
| def on_record(node): | |
| return isinstance(node, ast.Name) and aliases.get(node.id) == () | |
| for node in ast.walk(branch): | |
| if isinstance(node, (ast.Assign, ast.AugAssign)): | |
| targets = node.targets if isinstance(node, ast.Assign) else [node.target] | |
| for target in targets: | |
| if isinstance(target, ast.Attribute) and on_record(target.value) and ( | |
| watched is UNREADABLE or target.attr in roots | |
| or _clears(node.value)): | |
| return True | |
| elif (isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) | |
| and node.func.attr == "update" and on_record(node.func.value) | |
| and node.args and isinstance(node.args[0], ast.Dict)): | |
| for key, value in zip(node.args[0].keys, node.args[0].values): | |
| if isinstance(key, ast.Constant) and ( | |
| watched is UNREADABLE or key.value in roots or _clears(value)): | |
| return True | |
| return False | |
| def _raised_messages(scope): | |
| """The string literals inside every raise in this scope, long enough to | |
| name one check: `_()` and implicit concatenation are already one string.""" | |
| messages = set() | |
| for node in ast.walk(scope): | |
| if isinstance(node, ast.Raise): | |
| for inner in ast.walk(node): | |
| if isinstance(inner, ast.Constant) and isinstance(inner.value, str) \ | |
| and len(inner.value.strip()) >= 20: | |
| messages.add(inner.value.strip()) | |
| return messages | |
| def _check_helpers(scope, fields=None): | |
| """Names of the check-shaped methods this scope calls: _check_expiration_date, | |
| _validate_dates. A name is all there is to go on, so only those two words. | |
| With `fields`, only a call handed one of those fields counts: | |
| _check_expiration_date(self.expiration_date) checks the watched value, | |
| and a bare self._check_company() in the same onchange checks something | |
| else that write() may also happen to check. | |
| """ | |
| names = set() | |
| for node in ast.walk(scope): | |
| if not isinstance(node, ast.Call): | |
| continue | |
| name = _final_name(node.func) | |
| if not (name and ("check" in name.lower() or "validat" in name.lower())): | |
| continue | |
| if fields is not None and not any( | |
| isinstance(arg, ast.Attribute) and arg.attr in fields | |
| for arg in list(node.args) + [kw.value for kw in node.keywords]): | |
| continue | |
| names.add(name) | |
| return names | |
| def _server_methods(cls, fn): | |
| """Methods of this class the server runs whatever the client is: create(), | |
| write(), an @api.constrains method, and any method a field in the class | |
| names as its inverse=. Not action_post and friends: a check at posting | |
| lets a draft hold the value, which is the gap the rule is about.""" | |
| inverses = set() | |
| for node in ast.walk(cls): | |
| if isinstance(node, ast.Call): | |
| for keyword in node.keywords: | |
| if keyword.arg == "inverse" and isinstance(keyword.value, ast.Constant): | |
| inverses.add(keyword.value.value) | |
| return [m for m in cls.body | |
| if isinstance(m, (ast.FunctionDef, ast.AsyncFunctionDef)) and m is not fn | |
| and (m.name in ("create", "write") or m.name in inverses | |
| or _decorator_args(m, "api", "constrains") is not None)] | |
| def _server_repeats_the_check(tree, fn, watched): | |
| """Does a server-side method of the same class make the onchange's check? | |
| Two signals, and both have to be the SAME check rather than a similar one. | |
| The exact message the onchange raises, raised again: mrp.workorder's | |
| date_finished onchange refuses to unplan one work order, and _set_dates, | |
| the inverse of both date fields, raises the same sentence. Or the same | |
| check-shaped helper called from both: the API key wizard's onchange and | |
| its create() both call res.users.apikeys._check_expiration_date. | |
| Only the onchange's own class is searched, one call deep, because every | |
| yes here suppresses a finding. A check made at posting does not count | |
| (see _server_methods), and a paraphrased message is not the same check. | |
| """ | |
| cls = _owner_of(tree, fn) | |
| if cls is None or watched is UNREADABLE: | |
| return False | |
| messages = _raised_messages(fn) | |
| helpers = _check_helpers(fn, {w.split(".")[0] for w in watched}) | |
| if not messages and not helpers: | |
| return False | |
| for method in _server_methods(cls, fn): | |
| if messages & _raised_messages(method) or helpers & _check_helpers(method): | |
| return True | |
| return False | |
| def onchange_tries_to_enforce(code, facts=None): | |
| """odoo.onchange-is-not-validation | |
| An @api.onchange method that RAISES, or returns a warning while taking the | |
| value back -- it is trying to enforce an invariant the web client is the | |
| only thing that will apply. | |
| The regex this replaces read 400 characters after the decorator and could | |
| not see where the method ended. All three of its false positives are the | |
| same failure: a four-line onchange that assigns one field, followed by an | |
| action method that raises. `_onchange_variable_id` sets a target model and | |
| the UserError 200 characters below it belongs to `action_remap`. The | |
| window crossed the function boundary, and a regex has no way not to. | |
| Scoped to the decorated function, that shape is silent, and the four true | |
| positives -- an onchange that really does raise ValidationError, and ones | |
| that return the warning dict -- all stand. | |
| A constraint in the file clears it when it covers a field the onchange | |
| watches: an onchange beside a real @api.constrains on the same field is | |
| the faster feedback the rule explicitly allows. Until 2026-09-27 ANY | |
| constraint anywhere in the file cleared it, so a UNIQUE(reference) | |
| silenced an onchange that raised on credit_limit. _sql_constraints counts | |
| on 16 through 18 and on an unknown series, because it IS a constraint | |
| there, and not when the tree pins 19 or later, where the registry never | |
| creates it (_constrained_fields). | |
| Narrowing the guard alone added 36 findings and 11 read true, because the | |
| wide guard had been hiding a second problem: a returned warning dict was | |
| read as a refusal whatever it said. Most were notices -- "this reference | |
| already exists", "the rate is far from the previous one" -- that the user | |
| saves past in the form as easily as in an import, so nothing is enforced | |
| anywhere. A warning counts now only when its branch also takes the value | |
| back (_refuses_in_branch). Both together: OCA unchanged, OpenSPP +4 (3 | |
| true), core +11 (8 true) and -17, all 17 notices. ARCHITECTURE 3.67, 3.68. | |
| A check the server repeats in the same class also backs it: the same | |
| raised sentence, or the same check helper handed the watched field, in | |
| create(), write(), an inverse or an @api.constrains method | |
| (_server_repeats_the_check). Core 2,339 -> 2,338, the work order's | |
| unplanning check; nothing else moved. ARCHITECTURE 3.69. | |
| """ | |
| tree, _ = parse_python(code) | |
| if tree is None: | |
| return [] | |
| covered = _constrained_fields( | |
| tree, _series_at_least((facts or {}).get("odoo_series"), 19.0)) | |
| findings = [] | |
| for fn in _functions(tree): | |
| watched = _decorator_args(fn, "api", "onchange") | |
| if watched is None: | |
| continue | |
| if covered is EVERY_FIELD or (watched is UNREADABLE and covered): | |
| continue | |
| if watched is not UNREADABLE and any(w.split(".")[0] in covered | |
| for w in watched): | |
| continue | |
| if _server_repeats_the_check(tree, fn, watched): | |
| continue | |
| # Anchored on the decorator, not on the def below it. That is where | |
| # the rule is, where a reader looking the finding up will look, and | |
| # where the marker goes when somebody has already decided about it -- | |
| # a FunctionDef's own lineno is the `def`, one line past all three. | |
| at = fn.lineno | |
| for dec in fn.decorator_list: | |
| func = dec.func if isinstance(dec, ast.Call) else dec | |
| if isinstance(func, ast.Attribute) and func.attr == "onchange": | |
| at = dec.lineno | |
| break | |
| for node in ast.walk(fn): | |
| spot = None | |
| if isinstance(node, ast.Raise): | |
| exc = node.exc | |
| name = _final_name(exc.func) if isinstance(exc, ast.Call) \ | |
| else _final_name(exc) | |
| if name and ("Error" in name or "Warning" in name): | |
| spot = node.lineno | |
| elif isinstance(node, ast.Dict) and any( | |
| isinstance(key, ast.Constant) and key.value == "warning" | |
| for key in node.keys) and _refuses_in_branch(fn, node, watched): | |
| spot = node.lineno | |
| if spot is None: | |
| continue | |
| findings.append(( | |
| at, | |
| "@api.onchange '{0}' refuses a value, and onchange runs in " | |
| "the web client form only -- an import, an RPC call, a cron " | |
| "or a direct create() never triggers it, so the record enters " | |
| "the database unchallenged".format(fn.name), | |
| None)) | |
| break | |
| return findings | |
| # A query decides WHICH records you get to see, and dropping ir.rule widens | |
| # that set. Every one of the 6 hand-classified true positives is one of these. | |
| SUDO_QUERIES = ("search", "search_count", "search_read", "read_group") | |
| # These act on a recordset the caller already holds. Dropping ir.rule still | |
| # matters -- it governs write and read permission as well as visibility -- but | |
| # it widens what you may DO with a record rather than which records exist for | |
| # you, and nothing measured here has yet found a real one. | |
| SUDO_TOUCHES = ("create", "write", "unlink", "copy", "read") | |
| # The grade the write and escape clauses carry, per finding rather than per | |
| # rule. `unproven` is the corpus's word for a real defect nothing measured | |
| # here has met yet, and that is exactly their position: the read clause has | |
| # 6 hand-classified true positives across two trees, and these two have 0 in | |
| # 261,000 lines. Every one of the 23 they found was read, and every one was | |
| # an administrator by construction -- a demo-data seeder, an ACL-gated | |
| # accounting wizard -- or a framework model with no tenant boundary at all. | |
| # | |
| # Grading them rather than deleting them is the same call §3.46 made about | |
| # decimal-for-money: a clause the rule's own sentence asks for, measured at | |
| # zero, is reported and not trusted to stop a build. It costs the read | |
| # clause nothing, because a grade travels with the finding and not the rule. | |
| UNPROVEN = "unproven" | |
| def _escalates(call): | |
| """True when this `.sudo(...)` call turns record rules OFF. | |
| `sudo(False)` turns them back ON. It is how a method that must run under | |
| the caller's own rights says so, and odoo core does it 60 times -- | |
| `self = self.sudo(False)` in mail's `_check_attachments_access`, whose | |
| own docstring reads "This method relies on access rules/rights and | |
| therefore it should not be called from a sudo env." | |
| Reading it as an escalation reports the fix as the defect, which is worse | |
| than missing the defect: it is the one kind of finding that teaches a | |
| reader the checker cannot be trusted. An argument that is not a literal | |
| is unreadable here and counts as escalation, because `sudo(flag)` with a | |
| computed flag can be either and the finding should stand. | |
| """ | |
| if not (isinstance(call, ast.Call) and isinstance(call.func, ast.Attribute) | |
| and call.func.attr == "sudo"): | |
| return False | |
| given = list(call.args) + [k.value for k in call.keywords if k.arg == "flag"] | |
| for arg in given: | |
| if isinstance(arg, ast.Constant) and not arg.value: | |
| return False | |
| # `x.sudo(self.env.su)` sets su to whatever the environment already | |
| # has, so it cannot raise privilege above ambient -- it is how a | |
| # method hands a record back at the caller's own level. odoo core | |
| # writes it four times and once with the comment "Unsudo the invoice | |
| # after creation if not already sudoed". `sudo(self.env.su or | |
| # field.compute_sudo)` is NOT this: the `or` can turn it on, so only | |
| # the bare attribute is read as ambient. | |
| if isinstance(arg, ast.Attribute) and arg.attr == "su": | |
| return False | |
| return True | |
| def _under_sudo(node): | |
| """The receiver of `<x>.sudo().<method>(...)`, or None.""" | |
| if not (isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute)): | |
| return None | |
| inner = node.func.value | |
| if isinstance(inner, ast.Call) and _escalates(inner): | |
| return inner.func.value | |
| return None | |
| # Framework models odoo core gives no record rule, so sudo() on them has nothing | |
| # to drop. Measured over every ir.rule record in odoo core's XML: the only ir.* | |
| # models with one are ir.default, ir.embedded.actions, ir.filters, ir.ui.view and | |
| # ir.ui.view.custom. These four are the registry: no company, no owner, no | |
| # tenant. They are named rather than derived because a tree under review rarely | |
| # contains base, so whether base rules them is otherwise UNKNOWN -- which is how | |
| # three sudo() searches on ir.model stayed live on OCA after the AST move. | |
| # | |
| # ir.attachment is deliberately NOT here. It has no record rule either, but its | |
| # own _search filters by the caller's access to res_model unless the env is su, | |
| # so sudo() on it drops a real access check under another name. | |
| RULELESS_REGISTRY_MODELS = frozenset([ | |
| "ir.model", "ir.model.fields", "ir.model.data", "ir.module.module", | |
| ]) | |
| # A domain leaf ties a search to held records only through these. `!=`, `<` | |
| # and `not in` select everything else, which is the opposite of scoping. | |
| _HELD_OPERATORS = frozenset(["=", "in", "child_of"]) | |
| def _chain_escalates(node): | |
| """Does any call along this chain turn record rules off?""" | |
| while isinstance(node, (ast.Attribute, ast.Subscript, ast.Call)): | |
| if isinstance(node, ast.Call) and _escalates(node): | |
| return True | |
| node = node.func if isinstance(node, ast.Call) else node.value | |
| return False | |
| def _chain_calls(node, name): | |
| while isinstance(node, (ast.Attribute, ast.Subscript, ast.Call)): | |
| if isinstance(node, ast.Call) and _final_name(node.func) == name: | |
| return True | |
| node = node.func if isinstance(node, ast.Call) else node.value | |
| return False | |
| # Calls that hand back records the caller could already see, or a subset of | |
| # them. Anything else on a `self` chain is a helper whose body is not read here, | |
| # and it may sudo inside: a controller's `self._get_search_countries()` rooted at | |
| # `self` said nothing about whose rules produced the countries. | |
| _RULE_PRESERVING_CALLS = frozenset([ | |
| "filtered", "filtered_domain", "mapped", "sorted", "with_context", | |
| "with_company", "exists", "search", "grouped", | |
| ]) | |
| def _only_rule_preserving_calls(node): | |
| while isinstance(node, (ast.Attribute, ast.Subscript, ast.Call)): | |
| if isinstance(node, ast.Call) and _final_name(node.func) not in _RULE_PRESERVING_CALLS: | |
| return False | |
| node = node.func if isinstance(node, ast.Call) else node.value | |
| return True | |
| def _held_recordset(node, fn, depth=0): | |
| """Is this a recordset the caller obtained under its own record rules? | |
| `self` and anything reached from it without sudo: `self.filtered(...)`, | |
| `self.journal_id`, `self.env.user.partner_id`, a plain `search`. A name | |
| counts when every assignment to it does, and a loop variable when what it | |
| loops over does. `browse(ids)` does not count: browse applies no rule, so | |
| it holds whatever ids it was handed, and `env.ref(...)` does not either, | |
| for the same reason. Anything unreadable is not held, because every True | |
| here suppresses a finding. | |
| """ | |
| if node is None or depth > 4 or _chain_escalates(node) or _chain_calls(node, "browse") or _chain_calls(node, "ref"): | |
| return False | |
| root = _chain_root(node) | |
| if root == "self": | |
| return _only_rule_preserving_calls(node) | |
| if root == "request": | |
| # The user's own records and company, never an arbitrary lookup. | |
| text = ast.unparse(node) | |
| return text.startswith(("request.env.user", "request.env.company")) | |
| if root is None or fn is None: | |
| return False | |
| values = _assigned_values(fn, root) | |
| if values: | |
| return all(_held_recordset(v, fn, depth + 1) for v in values) | |
| for loop in ast.walk(fn): | |
| if isinstance(loop, (ast.For, ast.comprehension)) \ | |
| and isinstance(loop.target, ast.Name) and loop.target.id == root: | |
| return _held_recordset(loop.iter, fn, depth + 1) | |
| return False | |
| def _held_ids(node, fn, depth=0): | |
| """Ids of held records: `x.id`, `x.ids`, a list of them, or a name bound to one.""" | |
| if depth > 4: | |
| return False | |
| if isinstance(node, (ast.List, ast.Tuple)): | |
| real = [e for e in node.elts | |
| if not (isinstance(e, ast.Constant) and e.value in (False, None))] | |
| return bool(real) and all(_held_ids(e, fn, depth + 1) for e in real) | |
| if isinstance(node, ast.Attribute) and node.attr in ("id", "ids"): | |
| return _held_recordset(node.value, fn, depth + 1) | |
| if isinstance(node, ast.Name) and fn is not None: | |
| values = _assigned_values(fn, node.id) | |
| return bool(values) and all(_held_ids(v, fn, depth + 1) for v in values) | |
| return False | |
| def _tied_to_held_records(call, fn): | |
| """Does a domain leaf in this call's arguments pin it to held records? | |
| `('employee_id', 'in', self.ids)` returns the contracts of employees the | |
| caller can already see, and `('journal_id', '=', self.journal_id.id)` the | |
| moves of one journal, which belongs to one company. That is the rule's | |
| advice followed in a spelling the rule-field test cannot read: 18 of the | |
| 44 findings the global-rule test added on core (ARCHITECTURE 3.60, 3.64). | |
| A name passed as the domain is followed to what it was assigned. | |
| """ | |
| exprs = [] | |
| for arg in list(call.args) + [k.value for k in call.keywords]: | |
| exprs.append(arg) | |
| if isinstance(arg, ast.Name): | |
| exprs.extend(_assigned_values(fn, arg.id) or ()) | |
| for expr in exprs: | |
| for leaf in ast.walk(expr): | |
| if isinstance(leaf, (ast.Tuple, ast.List)) and len(leaf.elts) == 3 \ | |
| and isinstance(leaf.elts[0], ast.Constant) \ | |
| and isinstance(leaf.elts[0].value, str) \ | |
| and isinstance(leaf.elts[1], ast.Constant) \ | |
| and leaf.elts[1].value in _HELD_OPERATORS \ | |
| and _held_ids(leaf.elts[2], fn): | |
| return True | |
| return False | |
| # Prefix operators of an Odoo domain and how many terms each one takes. | |
| _DOMAIN_ARITY = {"|": 2, "&": 2, "!": 1} | |
| def _anded_leaves(elts): | |
| """Leaves ANDed at the top level of a domain list, or None when the list | |
| is not a complete domain. | |
| A domain is prefix notation: `'|', a, b` is a OR b, and the terms left | |
| over at the top are ANDed. Only a leaf reached through `&` alone is | |
| guaranteed to hold for every record returned; one under `|` or `!` is not. | |
| A list with an operator short of terms, `['|'] + domain`, is a fragment | |
| whose meaning depends on what it is glued to, so it answers None. | |
| """ | |
| def term(i): | |
| if i >= len(elts): | |
| return None, [] | |
| op = elts[i] | |
| if isinstance(op, ast.Starred): | |
| return None, [] # a spliced list could be a fragment | |
| if isinstance(op, ast.Constant) and op.value in _DOMAIN_ARITY: | |
| j, kept = i + 1, [] | |
| for _ in range(_DOMAIN_ARITY[op.value]): | |
| j, inner = term(j) | |
| if j is None: | |
| return None, [] | |
| kept.extend(inner) | |
| return j, kept if op.value == "&" else [] | |
| return i + 1, [op] | |
| i, leaves = 0, [] | |
| while i < len(elts): | |
| i, inner = term(i) | |
| if i is None: | |
| return None | |
| leaves.extend(inner) | |
| return leaves | |
| def _is_published_leaf(node): | |
| return isinstance(node, (ast.Tuple, ast.List)) and len(node.elts) == 3 \ | |
| and all(isinstance(e, ast.Constant) for e in node.elts) \ | |
| and [e.value for e in node.elts] == ["website_published", "=", True] | |
| def _domain_keeps_published(expr, fn, depth=0): | |
| """Does every record this domain can return have website_published = True? | |
| Followed through the spellings a controller builds a domain in: a list | |
| literal, `a + b`, `list(a)`, `a.copy()`, and a name every plain assignment | |
| to which qualifies. `domain += [...]` and `domain.append(leaf)` only AND | |
| more terms on, so they keep it, unless what they add is a fragment; an | |
| `insert` can put an operator in front of the leaf, so it does not. | |
| Anything unreadable answers False, because every True suppresses a finding. | |
| """ | |
| if depth > 4: | |
| return False | |
| if isinstance(expr, (ast.List, ast.Tuple)): | |
| leaves = _anded_leaves(expr.elts) | |
| return leaves is not None and any(_is_published_leaf(x) for x in leaves) | |
| if isinstance(expr, ast.BinOp) and isinstance(expr.op, ast.Add): | |
| sides = (expr.left, expr.right) | |
| if any(isinstance(s, (ast.List, ast.Tuple)) and _anded_leaves(s.elts) is None | |
| for s in sides): | |
| return False | |
| return any(_domain_keeps_published(s, fn, depth + 1) for s in sides) | |
| if isinstance(expr, ast.Call): | |
| if isinstance(expr.func, ast.Name) and expr.func.id == "list" \ | |
| and len(expr.args) == 1: | |
| return _domain_keeps_published(expr.args[0], fn, depth + 1) | |
| if isinstance(expr.func, ast.Attribute) and expr.func.attr == "copy" \ | |
| and not expr.args: | |
| return _domain_keeps_published(expr.func.value, fn, depth + 1) | |
| return False | |
| if not isinstance(expr, ast.Name) or fn is None \ | |
| or _assigned_values(fn, expr.id) is None: | |
| return False | |
| plain = [] | |
| for node in ast.walk(fn): | |
| if isinstance(node, ast.Assign) and any( | |
| isinstance(t, ast.Name) and t.id == expr.id for t in node.targets): | |
| plain.append(node.value) | |
| elif isinstance(node, ast.AugAssign) and isinstance(node.target, ast.Name) \ | |
| and node.target.id == expr.id: | |
| added = node.value | |
| if not isinstance(node.op, ast.Add) or ( | |
| isinstance(added, (ast.List, ast.Tuple)) | |
| and _anded_leaves(added.elts) is None): | |
| return False | |
| elif isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) \ | |
| and isinstance(node.func.value, ast.Name) \ | |
| and node.func.value.id == expr.id: | |
| if node.func.attr == "insert": | |
| return False | |
| if node.func.attr in ("append", "extend") and any( | |
| isinstance(a, ast.Constant) for a in node.args): | |
| return False | |
| return bool(plain) and all(_domain_keeps_published(v, fn, depth + 1) | |
| for v in plain) | |
| def _public_website_route(fn): | |
| """Is this function an @http.route(..., auth='public', website=True)?""" | |
| if not isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef)): | |
| return False | |
| for dec in fn.decorator_list: | |
| if not isinstance(dec, ast.Call) or _final_name(dec.func) != "route": | |
| continue | |
| kw = {k.arg: k.value for k in dec.keywords if k.arg} | |
| auth, website = kw.get("auth"), kw.get("website") | |
| if isinstance(auth, ast.Constant) and auth.value == "public" \ | |
| and isinstance(website, ast.Constant) and website.value is True: | |
| return True | |
| return False | |
| def _published_on_a_public_page(call, fn): | |
| """Is this a public website page reading only what it publishes? | |
| website_customer's /customers and website_crm_partner_assign's /partners | |
| list, group and count published partners under sudo(), because the | |
| anonymous visitor's own rules would hide most of them. Publishing is the | |
| record owner's decision to show it to everybody, and on a model with | |
| website.published.multi.mixin, searching website_published = True inside a | |
| website request also applies the current website's domain | |
| (_search_website_published, website/models/mixins.py). 12 of the 44 false | |
| positives the global-rule test added on core were this (ARCHITECTURE 3.60, | |
| 3.65). | |
| Narrow on purpose: a read, the literal leaf ('website_published', '=', | |
| True) ANDed at the top of the domain, inside a route that is both public | |
| and a website page. A published flag compared to anything but True, an | |
| `is_published` leaf, which applies no website, or the same search in a | |
| model method, stays a finding: website_sale's set_sequence_up is that. | |
| """ | |
| if call.func.attr not in SUDO_QUERIES or not _public_website_route(fn): | |
| return False | |
| domain = call.args[0] if call.args else next( | |
| (k.value for k in call.keywords if k.arg == "domain"), None) | |
| return domain is not None and _domain_keeps_published(domain, fn) | |
| def sudo_drops_record_rules(code, facts=None): | |
| """odoo.sudo-bypasses-record-rules | |
| Three clauses, because the rule has three: scope sudo to the narrowest | |
| operation, re-apply company filtering by hand, and never pass a sudo | |
| recordset onward. | |
| 1. a read under sudo -- search, search_read, read_group, read | |
| 2. a write under sudo -- create, write, unlink, copy | |
| 3. a sudo recordset bound to a name and then handed to something else | |
| Only the first was checked. The `--harvest` clustering says what that | |
| cost: `.sudo(` is the largest miss cluster in the corpus's recall | |
| measurement at 70 missed against 6 caught, and 47 of those 70 are files | |
| where a checker fired somewhere else -- a rule aimed wrong rather than a | |
| rule that is missing. Counted on the OpenSPP tree, the clause that was | |
| checked has 116 call sites and the clause that was not has 136, so the | |
| unchecked half was the larger one. One of the misses the README names, | |
| *"remove sudo() when importing record"*, is a create the search-only | |
| checker could not see at all. | |
| ir.rule governs read, write, create and unlink alike, so the same | |
| record-rule trigger gates every clause: where the tree declares no rule | |
| that filters this model, sudo drops nothing and there is no finding. The | |
| trigger only ever SUPPRESSES. A model the walk could not name, a tree that | |
| was not walked, a partial walk -- each reads UNKNOWN and the finding | |
| stands, because "no rule was found" and "nobody looked" must not be the | |
| same answer on a critical rule. `nexa check` on one file therefore behaves | |
| exactly as it did before triggers existed. | |
| Clause 3 fires at the binding, not at the use. `partners = | |
| self.env['res.partner'].sudo()` is a recordset carrying an environment | |
| with su=True, and every call made on it downstream inherits that -- which | |
| is the whole reason the rule's sentence says not to pass one on. Reported | |
| where the escape happens, once, rather than at each later read. | |
| What it over-reported, measured: 14 of the 20 surviving read findings | |
| constrained the search to records the caller had already identified -- a | |
| group id, a partner id, an area -- which is the rule's advice followed in a | |
| spelling the rule did not recognise. Three predicates now read that | |
| without evaluating the domain: the model's own rule fields (see rescoped), | |
| a leaf tying the search to ids of records the caller holds under its own | |
| rules (_tied_to_held_records, ARCHITECTURE 3.64), and a public website | |
| page reading only published records (_published_on_a_public_page, 3.65). | |
| None touches the six OpenSPP true positives, which are unscoped sweeps. | |
| """ | |
| tree, _ = parse_python(code) | |
| if tree is None: | |
| return [] | |
| classes = [n for n in ast.walk(tree) if isinstance(n, ast.ClassDef)] | |
| constants = _module_string_constants(tree) | |
| def rescoped(node, model): | |
| # Did this call re-apply the scoping the rule it bypasses is about? | |
| # | |
| # The old test was the literal word `company_id`, from the rule's own | |
| # correct sample. That is one spelling of one kind of rule, and the | |
| # docstring recorded what it cost: 14 of 20 surviving read findings | |
| # were scoped by a group, a partner or an area instead, and read as | |
| # unscoped. Every OpenSPP case model is governed by `case_worker_id` | |
| # or `case_id.case_worker_id` and carries no company rule at all. | |
| # | |
| # So the call is now compared against the fields the model's OWN | |
| # ir.rule filters on, which is not a guess about the domain -- it is | |
| # the domain, read off the tree that declares it. A create naming | |
| # `case_id` lands inside the caller's visibility because that is the | |
| # link the rule follows. company_id stays as the fallback for a model | |
| # whose rules this could not read, so nothing that was exempt before | |
| # stops being exempt now. | |
| # The ARGUMENTS only. Unparsing the whole call includes the receiver, | |
| # and the model name is part of it: `spp.group.membership` contains | |
| # the word `group`, which is one of that model's own rule fields, so | |
| # reading the whole chain exempted an unscoped sweep of every | |
| # membership in the database -- a finding classified true by hand. | |
| # | |
| # An argument that is a bare name is followed to what it was assigned, | |
| # because `create(vals)` otherwise hides every field it sets behind | |
| # four letters. The trace is the same local one the other checkers | |
| # use, and a name it cannot resolve contributes nothing. | |
| pieces = [] | |
| for arg in list(node.args) + [k.value for k in node.keywords]: | |
| try: | |
| pieces.append(ast.unparse(arg)) | |
| except Exception: # pragma: no cover | |
| continue | |
| if isinstance(arg, ast.Name): | |
| for value in _assigned_values(owner.get(id(node)), arg.id) or (): | |
| try: | |
| pieces.append(ast.unparse(value)) | |
| except Exception: # pragma: no cover | |
| pass | |
| call_src = " ".join(pieces) | |
| if _tied_to_held_records(node, owner.get(id(node))) \ | |
| or _published_on_a_public_page(node, owner.get(id(node))): | |
| return True | |
| def named(name): | |
| return re.search(r"\b" + re.escape(name.split(".", 1)[0]) + r"\b", | |
| call_src) is not None | |
| # Every GLOBAL rule, not any rule. Odoo ANDs the rules that carry no | |
| # groups for everybody and ORs group rules per audience, so naming | |
| # one field of one rule re-applies that rule and no other. | |
| # website_sale's set_sequence_up names website_published, the field of | |
| # product.template's portal and public rule, and was exempt while the | |
| # global company rule was dropped. Where the model has no global rule | |
| # the old test stands: any rule's field, then company_id. | |
| required = [fields for is_global, fields in | |
| (context.model_rule_sets(facts, model) if model else ()) | |
| if is_global and fields] | |
| if required: | |
| return all(any(named(name) for name in fields) for fields in required) | |
| fields = context.model_rule_fields(facts, model) if model else () | |
| for name in fields or ("company_id",): | |
| if named(name): | |
| return True | |
| return "company_id" in call_src | |
| owner = _function_of(tree) | |
| def model_of(receiver): | |
| # `self.env['x'].sudo()` names its model in the chain. `Plan.sudo()` | |
| # does not, and every one of the write clause's first findings on | |
| # OpenSPP's demo generators was that shape -- `Plan = | |
| # self.env['spp.case.plan']` a few lines up, then fifteen writes on | |
| # the name. Unresolved, the trigger reads UNKNOWN and cannot suppress; | |
| # followed through the assignment, it can answer. Every assignment to | |
| # the name must agree, or the answer is still None. | |
| model = _env_model_of(receiver) | |
| if model is not None or not isinstance(receiver, ast.Name): | |
| return model | |
| if receiver.id == "self": | |
| # `self` is the model the enclosing class defines or extends, when | |
| # one name says which: tracking_manager's `self.sudo().search()` | |
| # inside `_inherit = "ir.model"` is a search on ir.model. | |
| around = [c for c in classes | |
| if c.lineno <= receiver.lineno <= getattr(c, "end_lineno", c.lineno)] | |
| if around: | |
| return _own_model_name(around[-1], constants) | |
| values = _assigned_values(owner.get(id(receiver)), receiver.id) | |
| if not values: | |
| values = [v for n in ast.walk(tree) if isinstance(n, ast.Assign) | |
| for t in n.targets if isinstance(t, ast.Name) | |
| and t.id == receiver.id for v in (n.value,)] | |
| found = {_env_model_of(v) for v in values} | |
| return found.pop() if len(found) == 1 and None not in found else None | |
| def ruled_out(receiver): | |
| model = model_of(receiver) | |
| if model in RULELESS_REGISTRY_MODELS: | |
| return True, model | |
| return context.model_is_record_ruled(facts, model) == context.ABSENT, model | |
| findings, seen = [], set() | |
| def report(line, text, grade=None): | |
| if line not in seen: | |
| seen.add(line) | |
| findings.append((line, text, grade)) | |
| for node in ast.walk(tree): | |
| receiver = _under_sudo(node) | |
| if receiver is None: | |
| continue | |
| method = node.func.attr | |
| if method not in SUDO_QUERIES and method not in SUDO_TOUCHES: | |
| continue | |
| absent, model = ruled_out(receiver) | |
| if absent: | |
| continue | |
| if rescoped(node, model): | |
| continue | |
| where = " on {0}".format(model) if model else "" | |
| if method in SUDO_QUERIES: | |
| report(node.lineno, | |
| "sudo().{0}(){1} runs with record rules off -- ir.rule " | |
| "filtering, including multi-company isolation, does not " | |
| "apply and the result looks correct in single-company " | |
| "testing".format(method, where)) | |
| else: | |
| report(node.lineno, | |
| "sudo().{0}(){1} acts with record rules off -- ir.rule " | |
| "governs write, create, unlink and read as well as " | |
| "visibility, so this reaches a record the caller's own " | |
| "rules would not let it touch".format(method, where), | |
| UNPROVEN) | |
| # Clause 3: a sudo recordset given a name, then handed on. | |
| # | |
| # The bare name only, as an argument or as the whole returned value. | |
| # `return partners.mapped('name')` hands back a list of strings and the | |
| # elevated environment does not travel with it, while | |
| # `return partners.filtered(...)` does -- telling those apart means | |
| # knowing which recordset methods return recordsets, so neither is | |
| # reported. Clause 3 under-reports by construction, which is the right | |
| # direction for a new clause on a critical rule: the search clause's | |
| # first version fired 106 times for 6 real ones, and a flood is how a | |
| # checker gets switched off. | |
| for fn in _functions(tree): | |
| escapes = set() | |
| for node in ast.walk(fn): | |
| if isinstance(node, ast.Call): | |
| if _under_sudo(node) is None: | |
| escapes.update(a.id for a in node.args | |
| if isinstance(a, ast.Name)) | |
| escapes.update(k.value.id for k in node.keywords | |
| if isinstance(k.value, ast.Name)) | |
| elif isinstance(node, ast.Return) and isinstance(node.value, ast.Name): | |
| escapes.add(node.value.id) | |
| for node in ast.walk(fn): | |
| if not isinstance(node, ast.Assign) or not isinstance(node.value, ast.Call): | |
| continue | |
| call = node.value | |
| if not _escalates(call): | |
| continue | |
| bound = [t for t in _targets(node)] | |
| # `self = self.sudo()` rebinds the method's own receiver, and | |
| # every ORM call below it in the method inherits the elevated | |
| # environment without naming sudo anywhere. It is the escape | |
| # clause at its widest -- the recordset is not passed on, it | |
| # replaces the thing the whole method is written against. | |
| # | |
| # This is the miss the README names by hand. OCA server-tools | |
| # cbefaa511485, "[FIX] excel_import_export: remove sudo() when | |
| # importing record", deletes exactly this line and says what it | |
| # cost: "finding multiple records across all companies during | |
| # import when they have same record name. Previously, importing | |
| # with sudo could bypass multi-company record rules." | |
| # | |
| # Three sites in 260,000 lines, so it is narrow as well as real. | |
| # The same commit ADDS `self.sudo().env["ir.model.data"]`, a | |
| # deliberate and narrow elevation, which is the evidence for not | |
| # also flagging that shape. | |
| rebind = isinstance(call.func.value, ast.Name) \ | |
| and call.func.value.id in bound | |
| if rebind: | |
| report(node.lineno, | |
| "'{0}' is rebound to its own sudo form, so every ORM " | |
| "call below this line in the method runs with record " | |
| "rules off -- including ones that name no sudo at all" | |
| .format(call.func.value.id)) | |
| continue | |
| if not any(name in escapes for name in bound): | |
| continue | |
| absent, model = ruled_out(call.func.value) | |
| if absent: | |
| continue | |
| where = " of {0}".format(model) if model else "" | |
| report(node.lineno, | |
| "this sudo recordset{0} is passed on -- the elevated " | |
| "environment travels with it, so every read and write made " | |
| "on it downstream also runs with record rules off" | |
| .format(where), UNPROVEN) | |
| return sorted(findings) | |
| def _env_model_of(node): | |
| """The model name at the head of a chained call, or None. | |
| Resolves `self.env["res.partner"]` and `env["res.partner"]` through any | |
| number of intervening attributes. A head that is a plain name -- the | |
| `Partner` in `Partner.sudo().search(...)` -- is not resolvable without | |
| following the assignment, so it returns None and the caller treats the | |
| record-rule question as unanswered rather than answered no. | |
| """ | |
| while isinstance(node, (ast.Call, ast.Attribute)): | |
| node = node.func if isinstance(node, ast.Call) else node.value | |
| if isinstance(node, ast.Subscript) and isinstance(node.slice, ast.Constant) \ | |
| and isinstance(node.slice.value, str): | |
| target = node.value | |
| name = target.attr if isinstance(target, ast.Attribute) else ( | |
| target.id if isinstance(target, ast.Name) else None) | |
| if name == "env": | |
| return node.slice.value | |
| return None | |
| CHECKS = { | |
| "handler_order_shadows_a_subclass": handler_order_shadows_a_subclass, | |
| "sudo_drops_record_rules": sudo_drops_record_rules, | |
| "float_equality_against_zero": float_equality_against_zero, | |
| "path_join_takes_an_external_name": path_join_takes_an_external_name, | |
| "float_coercion_of_external_money": float_coercion_of_external_money, | |
| "onchange_tries_to_enforce": onchange_tries_to_enforce, | |
| "check_company_missing_on_company_owned_relation": | |
| check_company_missing_on_company_owned_relation, | |
| "commit_inside_a_loop": commit_inside_a_loop, | |
| "depends_names_unread_fields": depends_names_unread_fields, | |
| "sql_built_by_string_formatting": sql_built_by_string_formatting, | |
| "constrains_lists_every_field_read": constrains_lists_every_field_read, | |
| "constrains_declares_a_dotted_name": constrains_declares_a_dotted_name, | |
| "depends_covers_compute_reads": depends_covers_compute_reads, | |
| "ondelete_leaves_a_dangling_read": ondelete_leaves_a_dangling_read, | |
| "prefer_sql_constraint_applies": prefer_sql_constraint_applies, | |
| "parent_can_form_a_cycle": parent_can_form_a_cycle, | |
| "editable_record_without_noupdate": editable_record_without_noupdate, | |
| "naive_time_leaves_or_dates_a_day": naive_time_leaves_or_dates_a_day, | |
| "legacy_sql_constraints_on_19": legacy_sql_constraints_on_19, | |
| } | |
| # Checkers that read whole-tree facts as well as the file in front of them. A | |
| # set rather than signature introspection: which checkers are conditional on the | |
| # tree is a design fact worth being able to read off one line, and a checker that | |
| # grew a parameter by accident would otherwise start receiving one silently. | |
| CONTEXT_CHECKS = frozenset([ | |
| "ondelete_leaves_a_dangling_read", | |
| "check_company_missing_on_company_owned_relation", | |
| # Reads models_record_ruled: whether sudo drops anything depends on whether | |
| # the tree declares a rule that filters. | |
| "sudo_drops_record_rules", | |
| # Reads odoo_series: whether a _sql_constraints block is a constraint at | |
| # all depends on which Odoo is going to load it. | |
| "prefer_sql_constraint_applies", | |
| # Reads odoo_series, and whether a tree was walked at all: whether a | |
| # _sql_constraints block exists in the database depends on the series. | |
| "legacy_sql_constraints_on_19", | |
| # Reads odoo_series: a _sql_constraints block backs an onchange only on a | |
| # series that creates it. | |
| "onchange_tries_to_enforce", | |
| # Reads the ORM facts: whether a compute's field is stored, which | |
| # @api.depends Odoo merges into the method, and what a declared computed | |
| # field depends on in turn. A gap checker as well. | |
| "depends_covers_compute_reads", | |
| ]) | |
| # Checkers that can meet input they cannot decide about -- a decorator whose | |
| # arguments are computed -- and say so instead of guessing. Same reasoning as | |
| # CONTEXT_CHECKS: which checkers have a second channel is a design fact, and one | |
| # that grew a parameter by accident should not start receiving one silently. | |
| GAP_CHECKS = frozenset([ | |
| "constrains_lists_every_field_read", | |
| "constrains_declares_a_dotted_name", | |
| "depends_covers_compute_reads", | |
| ]) | |
| def run_check(name, code, facts=None, gaps=None): | |
| """Findings from one named checker. Unknown name raises, never passes. | |
| A renamed checker that silently returned 'clean' would turn a critical rule | |
| into a permanent false pass, which is the exact failure the corpus exists to | |
| prevent. corpus.py validate calls this for every rule carrying a | |
| semantic_check, so a typo fails the build. | |
| `gaps` collects what a checker could not decide -- a method whose decorator | |
| is computed rather than written out. Not findings: nothing is wrong with | |
| @api.constrains(*FIELDS), it is simply outside a static pass. None means the | |
| caller is not collecting them. | |
| `facts` is context.tree_facts for the tree this file came from, or None. The | |
| checkers in CONTEXT_CHECKS read a trigger off it; None means no tree was | |
| walked and every trigger reads UNKNOWN, which leaves those checkers behaving | |
| exactly as they did before triggers existed. | |
| Returns (line, message, grade) throughout. `grade` is a likelihood the | |
| finding earned on its own evidence -- a trigger that fired -- or None to keep | |
| the grade its rule already carries. Normalised here so the six checkers with | |
| no opinion about it do not each have to return a third element saying so. | |
| """ | |
| if name not in CHECKS: | |
| raise KeyError("no semantic checker named '{0}'".format(name)) | |
| if name in CONTEXT_CHECKS and name in GAP_CHECKS: | |
| found = CHECKS[name](code, facts, gaps=gaps) | |
| elif name in CONTEXT_CHECKS: | |
| found = CHECKS[name](code, facts) | |
| elif name in GAP_CHECKS: | |
| found = CHECKS[name](code, gaps=gaps) | |
| else: | |
| found = CHECKS[name](code) | |
| return [hit if len(hit) == 3 else (hit[0], hit[1], None) for hit in found] | |
| # ------------------------------------------------- imports and symbol checks | |
| # Symbols a generated Odoo or Python snippet uses constantly and cannot use | |
| # without an import. Curated rather than derived: a general undefined-name pass | |
| # over a method-shaped fragment flags the surrounding class it was lifted from | |
| # and buries the real finding under it. | |
| NEEDS_IMPORT = { | |
| "ValidationError": "odoo.exceptions", | |
| "UserError": "odoo.exceptions", | |
| "AccessError": "odoo.exceptions", | |
| "MissingError": "odoo.exceptions", | |
| "api": "odoo", | |
| "fields": "odoo", | |
| "models": "odoo", | |
| "float_compare": "odoo.tools", | |
| "float_is_zero": "odoo.tools", | |
| "float_round": "odoo.tools", | |
| "timedelta": "datetime", | |
| "Decimal": "decimal", | |
| "relativedelta": "dateutil.relativedelta", | |
| } | |
| def imported_names(tree): | |
| """Every name the module binds by importing it, aliases resolved.""" | |
| names = set() | |
| for node in ast.walk(tree): | |
| if isinstance(node, ast.Import): | |
| for alias in node.names: | |
| names.add(alias.asname or alias.name.split(".")[0]) | |
| elif isinstance(node, ast.ImportFrom): | |
| for alias in node.names: | |
| names.add(alias.asname or alias.name) | |
| return names | |
| def imported_modules(tree): | |
| """Every module a tree reaches for, by its real name, aliases seen through. | |
| Deliberately not imported_names(). That one answers "what does this module | |
| BIND", which is what resolving an undefined name needs. An allowlist needs | |
| "what does this REACH FOR", and the two disagree in both directions: | |
| from decimal import Decimal binds Decimal, reaches decimal | |
| import os as math binds math, reaches os | |
| A relative import resolves to a package this snippet is not part of, so it | |
| is reported as "." and can never match an allowlist entry. | |
| """ | |
| modules = set() | |
| for node in ast.walk(tree): | |
| if isinstance(node, ast.Import): | |
| for alias in node.names: | |
| modules.add(alias.name.split(".")[0]) | |
| elif isinstance(node, ast.ImportFrom): | |
| if node.level: | |
| modules.add(".") | |
| elif node.module: | |
| modules.add(node.module.split(".")[0]) | |
| return modules | |
| def bound_names(tree): | |
| """Names bound by assignment, def, class, except, or a for/with target. | |
| A snippet that defines its own 'fields' helper is not missing an import. | |
| """ | |
| names = set() | |
| for node in ast.walk(tree): | |
| if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): | |
| names.add(node.name) | |
| elif isinstance(node, ast.Name) and isinstance(node.ctx, ast.Store): | |
| names.add(node.id) | |
| elif isinstance(node, ast.ExceptHandler) and node.name: | |
| names.add(node.name) | |
| elif isinstance(node, ast.arg): | |
| names.add(node.arg) | |
| return names | |
| def is_lifted_fragment(tree, notes): | |
| """True when the block is a method shown without the class it belongs to. | |
| WHY THIS GATES THE IMPORT CHECK: a fragment like that was never going to | |
| carry an import header. The imports live in the module the answer quoted | |
| from, which was not part of the answer. Reporting them as missing is the | |
| same category of noise that the undefined-name analysis already declines to | |
| report -- and it is worse than useless, because a medium finding is enough | |
| to turn a correct answer's verdict into WARN and bury whatever else the run | |
| found. | |
| Two signals, either one sufficient: parse_fragment had to repair the block | |
| to make it parse at all (notes), or a function takes self or cls, which no | |
| module-level function does. | |
| """ | |
| if notes: | |
| return True | |
| for node in ast.walk(tree): | |
| if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): | |
| args = list(node.args.posonlyargs) + list(node.args.args) | |
| if args and args[0].arg in ("self", "cls"): | |
| return True | |
| return False | |
| def missing_imports(tree): | |
| """Curated symbols used without an import, as (line, message). | |
| Catches the failure the prompt explicitly asks the model to avoid -- 'Code | |
| first: runnable, imports included' -- and which a 7B drops routinely once | |
| the answer is long enough for that instruction to fall out of attention. | |
| """ | |
| have = imported_names(tree) | bound_names(tree) | |
| findings, seen = [], set() | |
| for node in ast.walk(tree): | |
| if not isinstance(node, ast.Name) or not isinstance(node.ctx, ast.Load): | |
| continue | |
| if node.id in have or node.id in seen or node.id not in NEEDS_IMPORT: | |
| continue | |
| seen.add(node.id) | |
| findings.append((node.lineno, | |
| "'{0}' is used but never imported (expected: from {1} import {0})" | |
| .format(node.id, NEEDS_IMPORT[node.id]))) | |
| return findings | |
| def unresolvable_imports(tree): | |
| """Imported modules that are not installed here, as (line, message). | |
| Implementation verification rather than plausibility: find_spec asks the | |
| interpreter whether the module actually exists on this machine, which is a | |
| different and stronger question than whether the import line looks | |
| reasonable. | |
| odoo is skipped. It is absent from this environment by design -- Nexa runs | |
| beside Odoo projects, not inside one -- so reporting it would be noise on | |
| every single Odoo answer. | |
| """ | |
| findings = [] | |
| for node in ast.walk(tree): | |
| if isinstance(node, ast.Import): | |
| modules = [a.name for a in node.names] | |
| elif isinstance(node, ast.ImportFrom): | |
| modules = [node.module] if node.module and not node.level else [] | |
| else: | |
| continue | |
| for module in modules: | |
| root = module.split(".")[0] | |
| if root in ("odoo", "openerp"): | |
| continue | |
| try: | |
| found = importlib.util.find_spec(root) is not None | |
| except (ImportError, ValueError, ModuleNotFoundError): | |
| found = False | |
| if not found: | |
| findings.append((node.lineno, | |
| "module '{0}' is not installed here, so the import " | |
| "cannot be verified".format(root))) | |
| return findings | |
| # ------------------------------------------------------- comment stripping | |
| def _docstring_starts(code): | |
| """Character offsets where a docstring literal opens. | |
| A docstring is prose, and a regex cannot tell prose from code -- which is | |
| the whole reason comments are stripped before matching. String literals | |
| were left standing, and classifying 50 findings by hand found the cost: | |
| two of them were `eval()` named in a docstring in a sentence arguing | |
| AGAINST using it, and a `datetime.now()` inside a doctest example. Both | |
| read as live code. | |
| Only docstrings, not every string. A triple-quoted SQL query is data the | |
| checkers have a legitimate interest in, and `_SQL_COLUMNS` reads one. The | |
| statement-level bare string an AST calls a docstring is the one place the | |
| file is guaranteed to be talking ABOUT code rather than in it. | |
| Offsets, not spans: the scanner already knows how to find a literal's end, | |
| and asking it to do that is one way of measuring the string rather than | |
| two. `col_offset` is a byte offset, which parts company with a character | |
| offset on a line holding non-ASCII -- so the result is checked against the | |
| quote it claims to point at and dropped when it does not, leaving that | |
| docstring intact rather than blanking the wrong run of characters. | |
| """ | |
| tree, _ = parse_python(code) | |
| if tree is None: | |
| return frozenset() | |
| line_start, offset = [], 0 | |
| for row in code.splitlines(True): | |
| line_start.append(offset) | |
| offset += len(row) | |
| starts = set() | |
| holders = (ast.Module, ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef) | |
| for node in ast.walk(tree): | |
| if not isinstance(node, holders): | |
| continue | |
| body = getattr(node, "body", None) | |
| if not body: | |
| continue | |
| first = body[0] | |
| if not (isinstance(first, ast.Expr) and isinstance(first.value, ast.Constant) | |
| and isinstance(first.value.value, str)): | |
| continue | |
| lineno = getattr(first.value, "lineno", 0) | |
| if not 0 < lineno <= len(line_start): | |
| continue | |
| at = line_start[lineno - 1] + first.value.col_offset | |
| # The guard the docstring above describes. A prefix of anything but | |
| # indentation is possible -- `def f(): "doc"` -- but it is ASCII in | |
| # every spelling that reaches here, so a miss means the offset is | |
| # wrong and the safe answer is to leave the literal alone. | |
| if at < len(code) and code[at] in ("'", '"'): | |
| starts.add(at) | |
| return frozenset(starts) | |
| def _blanked(text): | |
| """A string literal with its contents emptied, quotes and shape intact. | |
| Same length and the same newlines, so the line AND column a finding | |
| reports are the ones a person reading the file will see, and so a | |
| docstring that is the whole body of a function still leaves that function | |
| syntactically intact for anything that parses the stripped view. | |
| """ | |
| if len(text) < 2: | |
| return text | |
| quote = 3 if text[:3] in ("'''", '"""') else 1 | |
| head, body, tail = text[:quote], text[quote:-quote or None], text[-quote:] | |
| if len(text) < quote * 2: | |
| return text | |
| return head + "".join("\n" if c == "\n" else " " for c in body) + tail | |
| def strip_comments(code, kind="python"): | |
| """Code with its comments and docstrings removed, strings left intact. | |
| WHY: a violation_pattern and a satisfied_pattern are regexes, and a regex | |
| cannot tell code from prose. The adversarial counterexample probe put a | |
| number on what that costs -- all 12 satisfied_patterns in the corpus could be | |
| cleared by a COMMENT quoting the fix rather than applying it. Four of those | |
| were critical rules, including odoo.onchange-is-not-validation: write the | |
| onchange, add '# _sql_constraints = [', and the checker reported clean while | |
| the bug shipped. | |
| It cuts the other way too. A commented-out @api.onchange was being reported | |
| as a live violation, which is the false positive that teaches people to | |
| ignore a checker. | |
| A docstring is the same argument wearing a different quote. It was left | |
| standing here until classifying 50 findings by hand turned up two that were | |
| prose read as code, so docstrings are now emptied and left in place -- see | |
| _docstring_starts for why only docstrings and not every literal. | |
| Newlines inside removed comments are preserved so that multi-line patterns | |
| still see the line structure they were written against. | |
| """ | |
| out = [] | |
| i, n = 0, len(code) | |
| triple = ("'''", '"""') | |
| docstrings = _docstring_starts(code) if kind == "python" else frozenset() | |
| while i < n: | |
| ch = code[i] | |
| if kind == "python" and code[i:i + 3] in triple: | |
| marker = code[i:i + 3] | |
| end = code.find(marker, i + 3) | |
| end = n if end < 0 else end + 3 | |
| out.append(_blanked(code[i:end]) if i in docstrings else code[i:end]) | |
| i = end | |
| continue | |
| if ch in ("'", '"') or (kind == "js" and ch == "`"): | |
| j = i + 1 | |
| while j < n: | |
| if code[j] == "\\": | |
| j += 2 | |
| continue | |
| if code[j] == ch: | |
| j += 1 | |
| break | |
| if code[j] == "\n" and kind == "python": | |
| break # unterminated single-quoted string | |
| j += 1 | |
| out.append(_blanked(code[i:j]) if i in docstrings else code[i:j]) | |
| i = j | |
| continue | |
| # XML comments are stripped for both kinds: an Odoo answer mixes Python | |
| # and view XML in the same block often enough to matter. | |
| if code[i:i + 4] == "<!--": | |
| end = code.find("-->", i + 4) | |
| end = n if end < 0 else end + 3 | |
| out.append("\n" * code[i:end].count("\n")) | |
| i = end | |
| continue | |
| if kind == "python" and ch == "#": | |
| end = code.find("\n", i) | |
| i = n if end < 0 else end | |
| continue | |
| if kind == "js" and code[i:i + 2] == "//": | |
| end = code.find("\n", i) | |
| i = n if end < 0 else end | |
| continue | |
| if kind == "js" and code[i:i + 2] == "/*": | |
| end = code.find("*/", i + 2) | |
| end = n if end < 0 else end + 2 | |
| out.append("\n" * code[i:end].count("\n")) | |
| i = end | |
| continue | |
| out.append(ch) | |
| i += 1 | |
| return "".join(out) | |
| def looks_like_markup(code): | |
| """Is this a view file rather than source? Decides the comment syntax a | |
| mutation may safely insert.""" | |
| stripped = code.lstrip() | |
| return stripped.startswith("<") and not stripped.startswith("<<") | |
| # ------------------------------------------------ JavaScript / TypeScript | |
| _JS_OPEN = {"(": ")", "[": "]", "{": "}"} | |
| _JS_CLOSE = {")": "(", "]": "[", "}": "{"} | |
| # A slash starts a regex literal only where a value cannot already be complete. | |
| # The standard heuristic; wrong for a few legal programs, and documented as a | |
| # heuristic rather than sold as a parser. | |
| _REGEX_OK_AFTER = set("(,=:[!&|?{};+-*%~^<>") | {"\n", "\r", ""} | |
| # The other half of the same heuristic, and the half this was missing. A slash | |
| # after one of these words is a regex too, because none of them can END a value: | |
| # `return /re/`, `case /re/`, `typeof /re/`. Read as division instead, the regex | |
| # BODY was scanned as code, so a character class holding one bracket -- /[(]/, | |
| # /]/, /{/ -- was reported as an unbalanced delimiter in a file that is fine. | |
| # Only bites when the class is unbalanced, which is why /^[^@]+@[^@]+$/ survived | |
| # it: its brackets happen to pair. | |
| _REGEX_OK_AFTER_WORD = frozenset([ | |
| "return", "typeof", "instanceof", "in", "of", "new", "delete", "void", | |
| "throw", "case", "do", "else", "yield", "await", | |
| ]) | |
| def balance_js(code): | |
| """Delimiter balance for JS/TS, aware of strings, comments and templates. | |
| NOT a parser. There is no JavaScript grammar in the standard library, so | |
| this answers one question only: does every bracket that opens also close, in | |
| the right order, outside a string or comment? That catches the truncated | |
| answer and the run-together answer -- the two ways generated code most often | |
| fails to be code at all -- and says nothing about anything else. | |
| A caller wanting a real parse has to shell out to a toolchain, which this | |
| project deliberately does not require. The layer above records that gap as | |
| UNKNOWN instead of letting a balanced file read as a clean parse. | |
| """ | |
| stack, findings = [], [] | |
| i, line, n = 0, 1, len(code) | |
| while i < n: | |
| ch = code[i] | |
| nxt = code[i + 1] if i + 1 < n else "" | |
| if ch == "\n": | |
| line += 1 | |
| i += 1 | |
| continue | |
| if ch == "/" and nxt == "/": | |
| while i < n and code[i] != "\n": | |
| i += 1 | |
| continue | |
| if ch == "/" and nxt == "*": | |
| i += 2 | |
| while i < n - 1 and not (code[i] == "*" and code[i + 1] == "/"): | |
| if code[i] == "\n": | |
| line += 1 | |
| i += 1 | |
| i += 2 | |
| continue | |
| if ch in ("'", '"'): | |
| i, line = _skip_quoted(code, i, line, ch) | |
| continue | |
| if ch == "`": | |
| i, line = _skip_template(code, i, line) | |
| continue | |
| if ch == "/" and _starts_a_regex(code, i): | |
| i, line = _skip_regex(code, i, line) | |
| continue | |
| if ch in _JS_OPEN: | |
| stack.append((ch, line)) | |
| elif ch in _JS_CLOSE: | |
| if not stack: | |
| findings.append((line, "closing '{0}' with nothing open".format(ch))) | |
| elif stack[-1][0] != _JS_CLOSE[ch]: | |
| opener, opened_at = stack.pop() | |
| findings.append((line, "'{0}' closes a '{1}' opened on line {2}".format( | |
| ch, opener, opened_at))) | |
| else: | |
| stack.pop() | |
| i += 1 | |
| for opener, opened_at in stack: | |
| findings.append((opened_at, "'{0}' opened here is never closed".format(opener))) | |
| return findings | |
| def _starts_a_regex(code, i): | |
| """Is the slash at i opening a regex literal rather than dividing?""" | |
| token = _prev_significant(code, i) | |
| return token in _REGEX_OK_AFTER or token in _REGEX_OK_AFTER_WORD | |
| def _prev_significant(code, i): | |
| """The previous significant character -- or the whole WORD it ends. | |
| One character cannot tell `return` from an identifier ending in n, and | |
| that is exactly the difference between a regex and a division. Returning | |
| the word lets the caller ask the question it actually has. | |
| """ | |
| j = i - 1 | |
| while j >= 0 and code[j] in " \t": | |
| j -= 1 | |
| if j < 0: | |
| return "" | |
| if not (code[j].isalnum() or code[j] in "_$"): | |
| return code[j] | |
| end = j + 1 | |
| while j >= 0 and (code[j].isalnum() or code[j] in "_$"): | |
| j -= 1 | |
| return code[j + 1:end] | |
| def _skip_quoted(code, i, line, quote): | |
| i += 1 | |
| while i < len(code): | |
| if code[i] == "\\": | |
| i += 2 | |
| continue | |
| if code[i] == quote: | |
| return i + 1, line | |
| if code[i] == "\n": | |
| line += 1 # unterminated; balance reports the damage after | |
| i += 1 | |
| return i, line | |
| def _skip_template(code, i, line): | |
| """Template literals nest. Backtick-dollar-brace bodies are scanned as code, | |
| because a generated validation message routinely interpolates another | |
| template inside itself and the braces have to balance there too.""" | |
| i += 1 | |
| while i < len(code): | |
| if code[i] == "\\": | |
| i += 2 | |
| continue | |
| if code[i] == "`": | |
| return i + 1, line | |
| if code[i] == "\n": | |
| line += 1 | |
| if code[i] == "$" and i + 1 < len(code) and code[i + 1] == "{": | |
| depth = 1 | |
| i += 2 | |
| while i < len(code) and depth: | |
| if code[i] == "{": | |
| depth += 1 | |
| elif code[i] == "}": | |
| depth -= 1 | |
| elif code[i] == "\n": | |
| line += 1 | |
| elif code[i] == "`": | |
| i, line = _skip_template(code, i, line) | |
| continue | |
| i += 1 | |
| continue | |
| i += 1 | |
| return i, line | |
| def _skip_regex(code, i, line): | |
| i += 1 | |
| in_class = False | |
| while i < len(code): | |
| if code[i] == "\\": | |
| i += 2 | |
| continue | |
| if code[i] == "[": | |
| in_class = True | |
| elif code[i] == "]": | |
| in_class = False | |
| elif code[i] == "/" and not in_class: | |
| return i + 1, line | |
| elif code[i] == "\n": | |
| return i, line + 1 # a regex cannot span lines: it was division | |
| i += 1 | |
| return i, line | |
| if __name__ == "__main__": | |
| import json | |
| import sys | |
| if len(sys.argv) > 2 and sys.argv[1] == "check": | |
| print(json.dumps(run_check(sys.argv[2], sys.stdin.read()), indent=2)) | |
| else: | |
| print(__doc__) | |
| print("checkers: " + ", ".join(sorted(CHECKS))) | |