insoln · Oct 10, 2026 · Oct 10, 2026 · Oct 10, 2026
diff --git a/annet/rulebook/routeros/ethernet.py b/annet/rulebook/routeros/ethernet.py
@@ -0,0 +1,159 @@
+"""Identity-aware diff and patch logic for built-in RouterOS Ethernet interfaces."""
+
+from collections import OrderedDict as odict
+from dataclasses import dataclass
+from typing import Any
+
+from annet.annlib.rulebook.common import DiffDict, DiffItem, LogicResult, call_diff_logic
+from annet.types import Op
+from annet.vendors.library.routeros import RouterOSParseError, parse_attrs, quote, tokenize
+
+
+@dataclass(frozen=True)
+class EthernetSet:
+    identity: str
+    attrs: dict[str, str]
+
+
+# Values RouterOS drops from a compact export; an empty comment is how a comment is cleared.
+_EXPORT_DEFAULTS = {"comment": "", "disabled": "no", "mtu": "1500"}
+
+
+class EthernetSetParseError(ValueError):
+    """Raised when an Ethernet set command cannot be matched safely."""
+
+
+def parse_set(row: str, *, allow_positional: bool = False) -> EthernetSet:
+    """Parse an Ethernet set row and return its factory-port identity.
+
+    Both the spaced selector form ``[ find default-name=ether1 ]`` and the
+    compact generator form ``[find default-name=ether1]`` are accepted.
+    RouterOS compact exports may shorten an unchanged factory name to the
+    positional form ``set ether1``; callers may allow that form only while
+    parsing running configuration.
+    """
+    try:
+        tokens = tokenize(row)
+    except RouterOSParseError as exc:
+        raise EthernetSetParseError(str(exc)) from exc
+
+    if len(tokens) < 2 or tokens[0] != "set":
+        raise EthernetSetParseError(f"Expected a RouterOS Ethernet set row, got: {row!r}")
+
+    if tokens[1] == "[":
+        try:
+            close = tokens.index("]", 2)
+        except ValueError as exc:
+            raise EthernetSetParseError(f"Unclosed selector in RouterOS Ethernet row: {row!r}") from exc
+        selector = tokens[2:close]
+        if len(selector) != 2 or selector[0] != "find" or not selector[1].startswith("default-name="):
+            raise EthernetSetParseError(
+                f"Expected a single 'find default-name=...' selector in RouterOS Ethernet row: {row!r}"
+            )
+        identity = selector[1].split("=", 1)[1]
+        attrs_tokens = tokens[close + 1 :]
+    elif allow_positional and not tokens[1].isdigit():
+        identity = tokens[1]
+        attrs_tokens = tokens[2:]
+    else:
+        raise EthernetSetParseError(f"Expected a 'find default-name=...' selector in RouterOS Ethernet row: {row!r}")
+
+    if not identity:
+        raise EthernetSetParseError(f"Empty default-name identity in RouterOS Ethernet row: {row!r}")
+
+    try:
+        attrs = parse_attrs(attrs_tokens, row)
+    except RouterOSParseError as exc:
+        raise EthernetSetParseError(str(exc)) from exc
+    if "default-name" in attrs:
+        raise EthernetSetParseError(f"Conflicting default-name identity in RouterOS Ethernet row: {row!r}")
+    return EthernetSet(identity=identity, attrs=attrs)
+
+
+def _index(rows: odict[str, Any], side: str, *, allow_positional: bool) -> dict[str, tuple[str, EthernetSet]]:
+    indexed: dict[str, tuple[str, EthernetSet]] = {}
+    for row in rows:
+        parsed = parse_set(row, allow_positional=allow_positional)
+        if parsed.identity in indexed:
+            previous = indexed[parsed.identity][0]
+            raise EthernetSetParseError(
+                f"Duplicate default-name {parsed.identity!r} in {side} RouterOS Ethernet config: {previous!r}, {row!r}"
+            )
+        indexed[parsed.identity] = (row, parsed)
+    return indexed
+
+
+def _matches_desired(running: EthernetSet, desired: EthernetSet) -> bool:
+    """Compare the attributes explicitly owned by the desired row.
+
+    A compact RouterOS export omits attributes left at their default value.
+    Only platform-independent defaults are assumed for absent attributes; other
+    defaults differ between platforms (``disable-running-check`` on CHR, for
+    example), so an absent attribute without a known default still counts as drift.
+    """
+    return all(running.attrs.get(key, _EXPORT_DEFAULTS.get(key)) == value for key, value in desired.attrs.items())
+
+
+def diff(
+    old: odict[str, Any], new: odict[str, Any], diff_pre: odict[str, Any], _pops: tuple[str, ...] = (Op.AFFECTED,)
+) -> list[DiffItem]:
+    """Pair Ethernet set rows by factory default-name instead of command text."""
+    old_by_identity = _index(old, "running", allow_positional=True)
+    new_by_identity = _index(new, "desired", allow_positional=False)
+    old_positions = {identity: position for position, identity in enumerate(old_by_identity)}
+    items: list[tuple[int, DiffItem]] = []
+
+    for identity, (new_row, desired) in new_by_identity.items():
+        if identity in old_by_identity:
+            old_row, running = old_by_identity[identity]
+            changed = not _matches_desired(running, desired)
+            op = Op.MOVED if changed else _pops[-1]
+            children = call_diff_logic(
+                diff_pre[new_row]["subtree"], old.get(old_row, odict()), new[new_row], _pops + (op,)
+            )
+            position = old_positions[identity]
+            row = new_row if changed else old_row
+        else:
+            op = Op.ADDED
+            children = call_diff_logic(diff_pre[new_row]["subtree"], odict(), new[new_row], _pops + (op,))
+            position = len(old_by_identity) + len(items)
+            row = new_row
+
+        items.append(
+            (
+                position,
+                DiffItem(op=op, row=row, children=children, diff_pre=diff_pre[new_row]["match"]),
+            )
+        )
+
+    # A built-in interface that is absent from desired state is unmanaged, not removed.
+    for identity, (old_row, _) in old_by_identity.items():
+        if identity in new_by_identity:
+            continue
+        children = call_diff_logic(diff_pre[old_row]["subtree"], old[old_row], odict(), _pops + (Op.AFFECTED,))
+        items.append(
+            (
+                old_positions[identity],
+                DiffItem(op=Op.AFFECTED, row=old_row, children=children, diff_pre=diff_pre[old_row]["match"]),
+            )
+        )
+
+    items.sort(key=lambda pair: pair[0])
+    return [item for _, item in items]
+
+
+def _render(row: str) -> str:
+    parsed = parse_set(row)
+    parts = ["set", "[", "find", f"default-name={quote(parsed.identity)}", "]"]
+    for key, value in parsed.attrs.items():
+        parts.append(f"{key}={quote(value)}")
+    return " ".join(parts)
+
+
+def change(rule: dict[str, Any], key: tuple[str, ...], diff: DiffDict, **_: Any) -> LogicResult:
+    """Render changed desired rows as targeted in-place set commands."""
+    del rule, key
+    for item in diff[Op.ADDED] + diff[Op.MOVED]:
+        yield True, _render(item["row"]), item["children"]
+    for item in diff[Op.AFFECTED]:
+        yield True, item["row"], item["children"]
diff --git a/annet/rulebook/routeros/ignore_by_key.py b/annet/rulebook/routeros/ignore_by_key.py
@@ -1,11 +1,11 @@
-import shlex
 from collections import OrderedDict
 from typing import Any
 
 from contextlog import get_logger
 
 from annet.annlib.rulebook.common import DiffItem, default_diff
 from annet.annlib.types import Op
+from annet.vendors.library.routeros import parse_attrs, quote, tokenize
 
 
 def _normalize_command(
@@ -34,15 +34,25 @@ def _normalize_command(
 
     result_command = "add"
     try:
-        parts = shlex.split(command_line)
-
-        for part in parts[1:]:  # Skip the 'add' part
+        parts = tokenize(command_line)
+
+        start = 1
+        if parts and parts[0] == "set" and len(parts) > 1:
+            if parts[1] == "[":
+                start = parts.index("]", 2) + 1
+            elif "=" not in parts[1]:
+                start = 2
+        selector = parts[1:start]
+        if selector:
+            result_command += " " + " ".join(selector)
+        for part in parts[start:]:
             if "=" in part:
-                key_part = part.split("=", 1)[0]
-                # Skip if key is in ignore_keys
+                key_part, value = next(iter(parse_attrs([part], command_line).items()))
                 if key_part in ignore_keys:
                     continue
-            result_command += f" {part}"
+                result_command += f" {key_part}={quote(value)}"
+            else:
+                result_command += f" {quote(part)}"
 
         return result_command.strip()
 

diff --git a/annet/rulebook/routeros/remove_by_key.py b/annet/rulebook/routeros/remove_by_key.py
@@ -1,11 +1,32 @@
-import shlex
 from collections.abc import Iterator
 from ipaddress import ip_address
 from typing import Any
 
 from contextlog import get_logger
 
 from annet.annlib.types import Op
+from annet.vendors.library.routeros import RouterOSParseError, parse_attrs, quote, tokenize
+
+
+def _attrs(row: str) -> dict[str, str]:
+    parts = tokenize(row)
+    if not parts:
+        raise RouterOSParseError(f"Empty RouterOS command: {row!r}")
+    start = 1
+    if parts[0] == "set" and len(parts) > 1:
+        if parts[1] == "[":
+            try:
+                close = parts.index("]", 2)
+            except ValueError as exc:
+                raise RouterOSParseError(f"Unclosed selector in RouterOS row: {row!r}") from exc
+            # find inside a selector carries identity attributes; read them like shlex did
+            start = 2
+            parts = parts[:close] + parts[close + 1 :]
+        elif "=" not in parts[1]:
+            start = 2
+    # Keep the legacy permissive extraction of non-attribute words and duplicate keys.
+    attrs = [part for part in parts[start:] if "=" in part]
+    return {key.lower(): value for key, value in parse_attrs(attrs, row, strict=False).items()}
 
 
 def change(
@@ -28,37 +49,30 @@ def change(
         original_cmd = removed_cmd["row"]
 
         try:
-            parts = shlex.split(original_cmd)
+            params = _attrs(original_cmd)
         except Exception as e:
             get_logger().error("Command parsing failed: %s", e)
             continue
 
-        # Parse parameters into dictionary
-        params = {}
-        for part in parts[1:]:  # Skip the 'add' part
-            if "=" in part:
-                key_part, value = part.split("=", 1)
-                params[key_part.lower()] = value
-
         # Create remove command using match-case for cleaner logic
         match params:
             case {"name": name}:
-                yield True, f'remove name="{name}"', None
+                yield True, f"remove name={quote(name)}", None
 
             case {"peer": peer}:
                 if peer.startswith("*"):
                     yield True, "remove about", None
                 else:
-                    yield True, f'remove peer="{peer}"', None
+                    yield True, f"remove peer={quote(peer)}", None
 
             case {"host": host}:
-                yield False, f'remove host="{host}"', None
+                yield False, f"remove host={quote(host)}", None
 
             case {"action": action, "topics": topics}:
                 if action.startswith("*"):
                     yield True, "remove invalid", None
                 else:
-                    yield True, f'remove action="{action}" topics="{topics}"', None
+                    yield True, f"remove action={quote(action)} topics={quote(topics)}", None
 
             case {"address": address, "interface": interface}:
                 addr, sep, mask = address.partition("/")
@@ -67,24 +81,28 @@ def change(
                     mask = str(ip.max_prefixlen)
                     address = f"{addr}/{mask}"
                 if interface.startswith("*"):
-                    yield True, f'remove address="{address}"', None
+                    yield True, f"remove address={quote(address)}", None
                 else:
-                    yield True, f'remove address="{address}" interface="{interface}"', None
+                    yield True, f"remove address={quote(address)} interface={quote(interface)}", None
 
             case {"address": address}:
-                yield True, f'remove address="{address}"', None
+                yield True, f"remove address={quote(address)}", None
 
             case {"interface": interface, "list": list}:
                 if interface.startswith("*"):
-                    yield True, f'remove list="{list}"', None
+                    yield True, f"remove list={quote(list)}", None
                 else:
-                    yield True, f'remove interface="{interface}"', None
+                    yield True, f"remove interface={quote(interface)}", None
 
             case {"comment": comment, "dst-address": dst_address, "gateway": gateway}:
-                yield True, f'remove comment="{comment}" dst-address="{dst_address}" gateway="{gateway}"', None
+                yield (
+                    True,
+                    f"remove comment={quote(comment)} dst-address={quote(dst_address)} gateway={quote(gateway)}",
+                    None,
+                )
 
             case {"comment": comment, "dst-address": dst_address}:
-                yield True, f'remove comment="{comment}" dst-address="{dst_address}"', None
+                yield True, f"remove comment={quote(comment)} dst-address={quote(dst_address)}", None
 
             case {"disabled": disabled, "topics": topics}:
                 # remove all disabled topics
@@ -93,7 +111,7 @@ def change(
 
             case {"topics": topics}:
                 topics = topics.replace(",", ".")
-                yield True, f'remove topics~"{topics}"', None
+                yield True, f"remove topics~{quote(topics)}", None
 
             case {"disabled": _}:
                 # Always turn off www-ssl

diff --git a/annet/rulebook/texts/routeros.rul b/annet/rulebook/texts/routeros.rul
@@ -24,6 +24,8 @@ file
     set * %logic=annet.rulebook.routeros.file.change
 
 interface
+    ethernet
+        set ... %logic=annet.rulebook.routeros.ethernet.change %diff_logic=annet.rulebook.routeros.ethernet.diff
     gre
         add ~ %logic=annet.rulebook.routeros.remove_by_key.change
     l2tp-client