From c0dd45189bc702d7975f63f6de9f8de3c5814268 Mon Sep 17 00:00:00 2001 From: Thomas Mieslinger Date: Tue, 18 Aug 2026 15:07:56 +0200 Subject: [PATCH] fix(dim): make the ancestor query correct at prefix 0 instead of avoiding it _ancestors_noparent_condition() builds its WHERE clause by joining one term per prefix length above the block. At prefix 0 there is nothing above, so the join produced an empty string, which it then wrapped in parentheses and handed to the database as "()" -- a 1064 syntax error. That was worked around by giving each caller that could reach prefix 0 its own `if ip.prefix != 0` guard: in Ipblock._tree_update(), in _find_ipblock() and on the guess_function of ipblock_create(). The broken query stayed, waiting for the next caller. Fix the cause: an empty disjunction is false, so return false() and let callers get the empty result set that is the correct answer. The three guards then say nothing the query does not already say, and are removed. Verified behaviour-neutral: of the seven new tests, five pass both with and without this change -- they pin down exactly what the guards used to provide. The other two go at the query directly and are the ones that fail without it. Not addressed, found while reading: _find_ipblock() computes `status_str = ' or '.join(status)` unconditionally, but its `status` parameter defaults to None and ipblock_list() passes it through. Reachable over the API with a block that does not exist exactly; not reachable through ndcli, which never calls ipblock_list. --- dim-testsuite/tests/ancestors_test.py | 75 +++++++++++++++++++++++++++ dim/dim/models/ip.py | 16 +++--- dim/dim/rpc.py | 19 ++++--- 3 files changed, 94 insertions(+), 16 deletions(-) create mode 100644 dim-testsuite/tests/ancestors_test.py diff --git a/dim-testsuite/tests/ancestors_test.py b/dim-testsuite/tests/ancestors_test.py new file mode 100644 index 0000000..cc599c4 --- /dev/null +++ b/dim-testsuite/tests/ancestors_test.py @@ -0,0 +1,75 @@ +"""Ancestor lookup at prefix 0. + +A /0 has no block above it, so the ancestor query has to produce an empty +result. It used to build its WHERE clause by joining one term per prefix +length above the block -- an empty string at prefix 0 -- and wrapped that in +parentheses, which MariaDB rejected with a 1064 syntax error. Callers were +then given a `prefix != 0` guard each, which left the broken query in place +for anyone who added a caller later. + +These tests go at the query directly rather than through the API, so they fail +if the guards come back instead of the cause being fixed. +""" +from dim.ipaddr import IP +from dim.models import Ipblock, Layer3Domain +from tests.util import RPCTest + + +class AncestorsPrefixZeroTest(RPCTest): + def setUp(self): + RPCTest.setUp(self) + self.layer3domain = Layer3Domain.query.filter_by(name='default').one() + + def ancestors(self, cidr, include_self=False): + return Ipblock._ancestors_noparent(IP(cidr), self.layer3domain, + include_self=include_self) + + def test_v4_default_route_has_no_ancestors(self): + assert self.ancestors('0.0.0.0/0') == [] + + def test_v6_default_route_has_no_ancestors(self): + assert self.ancestors('::/0') == [] + + def test_default_route_finds_itself_with_include_self(self): + self.r.ipblock_create('0.0.0.0/0', status='Container', layer3domain='default') + found = self.ancestors('0.0.0.0/0', include_self=True) + assert [str(b.ip) for b in found] == ['0.0.0.0/0'] + + def test_default_route_is_an_ancestor_of_a_block_below_it(self): + self.r.ipblock_create('0.0.0.0/0', status='Container', layer3domain='default') + self.r.ipblock_create('10.0.0.0/8', status='Container', layer3domain='default') + assert [str(b.ip) for b in self.ancestors('10.0.0.0/8')] == ['0.0.0.0/0'] + + def test_default_route_becomes_the_parent_of_blocks_below_it(self): + '''_tree_update() has to see the /0 as a possible parent. + + This is what the guard in _tree_update() used to skip: it returned + parent = None for the /0 itself, which is correct, but only because the + query underneath would have crashed. + ''' + self.r.ipblock_create('0.0.0.0/0', status='Container', layer3domain='default') + self.r.ipblock_create('10.0.0.0/8', status='Container', layer3domain='default') + root = Ipblock.query_ip(IP('0.0.0.0/0'), self.layer3domain).one() + child = Ipblock.query_ip(IP('10.0.0.0/8'), self.layer3domain).one() + assert root.parent is None + assert child.parent is not None and child.parent.id == root.id + + def test_v4_and_v6_default_routes_do_not_see_each_other(self): + '''Both carry address 0 and prefix 0; only the version separates them.''' + self.r.ipblock_create('0.0.0.0/0', status='Container', layer3domain='default') + assert self.ancestors('::/0', include_self=True) == [] + + def test_container_layer3domain_is_guessed_for_a_default_route(self): + '''ipblock_create() passes its guess_function unconditionally now. + + With more than one layer3domain and none given, the guess finds no + parent for a /0 and the call has to fail with the regular DIM error -- + not with a SQL syntax error, and not by skipping the guess entirely. + ''' + self.r.layer3domain_create('other', 'vrf', rd='8560:2') + try: + self.r.ipblock_create('0.0.0.0/0', status='Container') + except Exception as e: + assert 'A layer3domain is needed' in str(e), str(e) + else: + raise AssertionError('expected the call to require a layer3domain') diff --git a/dim/dim/models/ip.py b/dim/dim/models/ip.py index 4d916cf..8ed482f 100644 --- a/dim/dim/models/ip.py +++ b/dim/dim/models/ip.py @@ -5,7 +5,7 @@ from sqlalchemy import Column, BigInteger, Integer, String, Numeric, TIMESTAMP, ForeignKey, UniqueConstraint from sqlalchemy.ext.compiler import compiles from sqlalchemy.orm import relationship, backref, validates, synonym -from sqlalchemy.sql import bindparam, or_, between, func, expression, text +from sqlalchemy.sql import bindparam, or_, between, func, expression, false, text from sqlalchemy.types import DateTime from dim import db @@ -432,10 +432,8 @@ def build_tree_parents(layer3domain, version): def _tree_update(self): db.session.flush() # we need self.id logging.debug('Updating tree for %s', self) - new_parent_id = None - if self.ip.prefix !=0: - ancestors = Ipblock._ancestors_noparent(self.ip, self.layer3domain) - new_parent_id = ancestors[0].id if ancestors else None + ancestors = Ipblock._ancestors_noparent(self.ip, self.layer3domain) + new_parent_id = ancestors[0].id if ancestors else None if self.parent_id != new_parent_id: self.parent_id = new_parent_id if not self.is_host: @@ -504,10 +502,16 @@ def _ancestors_noparent(ip, layer3domain, include_self=False): @staticmethod def _ancestors_noparent_condition(ip, include_self=False): + prefixes = range(ip.prefix + (1 if include_self else 0)) + if not prefixes: + # A /0 has nothing above it, so the disjunction below is empty -- + # which is false, not the syntactically broken '()' this used to + # emit. Callers get an empty result set, which is the right answer. + return false() # use raw sql because sqlalchemy is slow at building large queries f = " OR ".join(('address=%d AND prefix=%d' % (ip.address & ((2 ** ip.bits - 1) ^ (2 ** (ip.bits - prefix) - 1)), prefix) - for prefix in range(ip.prefix + (1 if include_self else 0)))) + for prefix in prefixes)) return text('(' + f + ')') @staticmethod diff --git a/dim/dim/rpc.py b/dim/dim/rpc.py index 3617c73..ea0bb61 100644 --- a/dim/dim/rpc.py +++ b/dim/dim/rpc.py @@ -425,7 +425,7 @@ def find_parent(): return parents[0].layer3domain layer3domain = _get_layer3domain_arg(layer3domain, options, - guess_function=find_parent if status == 'Container' and parse_ip(block_str).prefix !=0 else None) + guess_function=find_parent if status == 'Container' else None) ip = check_ip(parse_ip(block_str), layer3domain, options) ipblock = Ipblock.query_ip(ip, layer3domain).first() pool = self._can_change_ip(ipblock or ip, layer3domain=layer3domain) @@ -4186,15 +4186,14 @@ def _find_ipblock(ipblock, layer3domain, status=None): return block status_str = ' or '.join(status) # Try ancestors - if ip.prefix != 0: - parents = Ipblock._ancestors_noparent_query(ip, layer3domain) - if status: - parents = parents.join(IpblockStatus).filter(IpblockStatus.name.in_(status)) - parents = parents.all() - if parents: - Messages.warn('%s rounded to %s because no ipblock exists at %s with status %s' - % (ip, parents[0].ip, ip, status_str)) - return parents[0] + parents = Ipblock._ancestors_noparent_query(ip, layer3domain) + if status: + parents = parents.join(IpblockStatus).filter(IpblockStatus.name.in_(status)) + parents = parents.all() + if parents: + Messages.warn('%s rounded to %s because no ipblock exists at %s with status %s' + % (ip, parents[0].ip, ip, status_str)) + return parents[0] # Try descendants descendants = Ipblock.query.filter(inside(Ipblock.address, ip), Ipblock.version == ip.version,