fix: resolve skill_id from bus context in spec and legacy detach handlers

handle_register_template, handle_register_entity, handle_deregister_intent,
handle_deregister_entity, handle_deregister_skill and the legacy
handle_detach_skill trusted message.data['skill_id'] unconditionally, so a
payload disagreeing with message.context['skill_id'] silently acted under
another skill's name. The bus client twins ovos.skill.deregister into the
legacy detach_skill topic carrying the original context, so guarding only
the spec handler left the legacy twin as a bypass: on FakeBus with
ovos-bus-client 2.11.14a2, ovos.skill.deregister with data.skill_id=victim
and context.skill_id=attacker still removed the victim's intents.

Add module-level _skill_id_from_context(message, handler): when context
carries a skill_id it is used regardless of the payload (a mismatch is
logged at WARNING and never acted on); when context carries no skill_id
the payload value is trusted, which is where this helper diverges from the
ovos_adapt and ovos_padatious twins (those return None). Wire it into the
five spec handlers and into handle_detach_skill.

The legacy handle_detach_intent, twinned from ovos.intent.deregister with
the producer's context, compares the intent_name owner prefix against
context.skill_id and rejects a mismatch, with the same guard condition as
ovos_adapt and ovos_padatious: a truthy context skill_id AND a namespaced
intent_name. A bare unnamespaced name or an empty-string context keeps the
pre-spec detach.

Fail-before against origin/dev (tests kept, source reverted): 4 failed /
9 passed of the detach/deregister set; the bus-emitted deregister case
fails with [] != ['victim.skill:play_music']. After: 13 passed. Full suite
109 passed, 12 failed (on top of padacioso#102, whose prefix match
handle_detach_skill keeps); origin/dev in the same venv is 96 passed, 12 failed
with the identical failed set (ovoscope register_padatious_intent skill_id
kwarg drift, tracked in padacioso#97).

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

Author: JarbasAl

padacioso/opm.py

@@ -280,10 +280,27 @@ def __detach_intent(self, intent_name, lang=None):
     def handle_detach_intent(self, message):
         """Messagebus handler for detaching padacioso intent.
 
+        The bus-client twins ``ovos.intent.deregister`` into this legacy
+        topic, building ``intent_name`` (``<skill_id>:<name>``) from the
+        payload while forwarding the producer's original context, so the
+        prefix is compared against ``context["skill_id"]`` and a mismatch
+        is rejected. A bare, unnamespaced name or a message without a
+        context skill_id keeps the pre-spec behaviour.
+
         Args:
             message (Message): message triggering action
         """
-        self.__detach_intent(message.data.get('intent_name'))
+        intent_name = message.data.get('intent_name')
+        context_skill_id = message.context.get("skill_id") if message.context else None
+        if context_skill_id and intent_name and ":" in intent_name:
+            owner = intent_name.split(":", 1)[0]
+            if owner != context_skill_id:
+                LOG.warning(f"[handle_detach_intent] rejected: intent_name="
+                            f"{intent_name!r} does not belong to "
+                            f"message.context['skill_id']={context_skill_id!r} "
+                            f"(topic={message.msg_type})")
+                return
+        self.__detach_intent(intent_name)
 
     def __detach_entity(self, name, lang):
         """ Remove an entity.
@@ -300,10 +317,16 @@ def __detach_entity(self, name, lang):
     def handle_detach_skill(self, message):
         """Messagebus handler for detaching all intents for skill.
 
+        The bus-client twins ``ovos.skill.deregister`` into this legacy
+        topic with the producer's original context, so the skill_id is
+        resolved the same way as in ``handle_deregister_skill``.
+
         Args:
             message (Message): message triggering action
         """
-        skill_id = message.data['skill_id']
+        skill_id = _skill_id_from_context(message, "handle_detach_skill")
+        if not skill_id:
+            return
         prefix = skill_id + ":"
         remove_list = [i for i in self.registered_intents if i.startswith(prefix)]
         for i in remove_list:
@@ -450,7 +473,7 @@ def handle_register_template(self, message: Message):
         """
         topic = SpecMessage.INTENT_REGISTER_TEMPLATE.value
         data = message.data
-        skill_id = data.get("skill_id")
+        skill_id = _skill_id_from_context(message, "handle_register_template")
         intent_name = data.get("intent_name")
         samples = data.get("samples")
         if not samples:  # §6.3 malformed
@@ -494,7 +517,7 @@ def handle_register_entity(self, message: Message):
         """OVOS-INTENT-4 §7 — register an entity value-set hint."""
         topic = SpecMessage.ENTITY_REGISTER.value
         data = message.data
-        skill_id = data.get("skill_id")
+        skill_id = _skill_id_from_context(message, "handle_register_entity")
         entity_name = data.get("entity_name")
         samples = data.get("samples")
         if not samples:  # §7.2 malformed
@@ -535,7 +558,7 @@ def _intent_langs(self, message: Message) -> List[str]:
 
     def handle_deregister_intent(self, message: Message):
         """OVOS-INTENT-4 §8.2 — remove one intent (all langs if lang omitted)."""
-        skill_id = message.data.get("skill_id")
+        skill_id = _skill_id_from_context(message, "handle_deregister_intent")
         intent_name = message.data.get("intent_name")
         name = self._internal_name(skill_id, intent_name)
         self.__detach_intent(name)
@@ -544,7 +567,7 @@ def handle_deregister_intent(self, message: Message):
 
     def handle_deregister_entity(self, message: Message):
         """OVOS-INTENT-4 §8.3 — remove one entity (all langs if lang omitted)."""
-        skill_id = message.data.get("skill_id")
+        skill_id = _skill_id_from_context(message, "handle_deregister_entity")
         entity_name = message.data.get("entity_name")
         name = self._internal_name(skill_id, entity_name)
         for lang in self._intent_langs(message):
@@ -554,7 +577,7 @@ def handle_deregister_entity(self, message: Message):
 
     def handle_deregister_skill(self, message: Message):
         """OVOS-INTENT-4 §8.4 — remove everything owned by a skill_id."""
-        skill_id = message.data.get("skill_id")
+        skill_id = _skill_id_from_context(message, "handle_deregister_skill")
         if not skill_id:
             return
         prefix = skill_id + ":"
@@ -731,6 +754,36 @@ def shutdown(self):
         self.bus.remove(SpecMessage.INTENT_DISABLE.value, self.handle_disable_intent)
 
 
+def _skill_id_from_context(message: Message, handler: str) -> Optional[str]:
+    """Resolve the skill id a handler acts on from the bus context.
+
+    When ``message.context["skill_id"]`` is set it is used, and a payload
+    ``skill_id`` that differs from it is logged and ignored, so the handler
+    proceeds under the context's name. This matches ``ovos_adapt`` and
+    ``ovos_padatious``. When the context carries no skill_id, the payload
+    value is trusted; here the helper diverges from those twins, which
+    return ``None`` in that case.
+
+    Args:
+        message: the incoming bus message
+        handler: name of the calling handler, for the warning message
+
+    Returns:
+        The skill id from the context, or the payload's when the context
+        carries none.
+    """
+    context_skill_id = message.context.get("skill_id") if message.context else None
+    payload_skill_id = message.data.get("skill_id")
+    if context_skill_id is None:
+        return payload_skill_id
+    if payload_skill_id and payload_skill_id != context_skill_id:
+        LOG.warning(f"[{handler}] message.data['skill_id']={payload_skill_id!r} "
+                    f"differs from message.context['skill_id']="
+                    f"{context_skill_id!r} on {message.msg_type}; "
+                    f"using the context value")
+    return context_skill_id
+
+
 def _dealias_intent_name(name: Optional[str]) -> Optional[str]:
     """Fold the legacy ``<skill_id>:<file>.intent`` id onto the OVOS-INTENT-4
     canonical ``<skill_id>:<file>`` id.

test/test_pipeline.py

@@ -217,6 +217,142 @@ def test_legacy_still_works(self):
         intent = svc.calc_intent("hello there", "en-US")
         self.assertEqual(intent.name, "greet.skill:hello")
 
+    def test_mismatched_register_template_uses_context_skill_id(self):
+        # OVOS-INTENT-4 §3.2 — message.context is attached by the bus client
+        # and cannot be forged by a producer, so it is authoritative: a
+        # payload skill_id that disagrees with it is logged and ignored, and
+        # the registration proceeds under the CONTEXT skill_id (matches
+        # ovos_adapt/opm.py:73 and ovos_padatious's twin helper)
+        svc = self.get_service()
+        msg = Message(self.SpecMessage.INTENT_REGISTER_TEMPLATE.value, {
+            "skill_id": "payload.skill",
+            "intent_name": "play_music",
+            "lang": "en-US",
+            "samples": ["play {query}"],
+        }, {"skill_id": "context.skill"})
+        svc.handle_register_template(msg)
+        self.assertIn("context.skill:play_music",
+                      svc.containers["en-US"].intent_samples)
+        self.assertNotIn("payload.skill:play_music",
+                         svc.containers["en-US"].intent_samples)
+
+    def test_mismatched_deregister_skill_targets_context_skill_id(self):
+        # the resolved skill_id (context, not payload) is what gets
+        # deregistered: a mismatched ovos.skill.deregister acts on the
+        # CONTEXT skill's intents, leaving the payload-named skill's own
+        # intents untouched
+        svc = self.get_service()
+        svc.handle_register_template(Message(
+            self.SpecMessage.INTENT_REGISTER_TEMPLATE.value, {
+                "skill_id": "victim.skill", "intent_name": "play_music",
+                "lang": "en-US", "samples": ["play {query}"],
+            }))
+        svc.handle_deregister_skill(Message(
+            self.SpecMessage.SKILL_DEREGISTER.value,
+            {"skill_id": "victim.skill"},
+            {"skill_id": "attacker.skill"}))
+        self.assertIn("victim.skill:play_music",
+                      svc.containers["en-US"].intent_samples)
+        intent = svc.calc_intent("play jazz", "en-US")
+        self.assertEqual(intent.name, "victim.skill:play_music")
+
+    def _register_victim(self, svc):
+        svc.bus.emit(Message(
+            self.SpecMessage.INTENT_REGISTER_TEMPLATE.value, {
+                "skill_id": "victim.skill", "intent_name": "play_music",
+                "lang": "en-US", "samples": ["play {query}"],
+            }, {"skill_id": "victim.skill"}))
+        self.assertIn("victim.skill:play_music",
+                      svc.containers["en-US"].intent_samples)
+
+    def test_bus_emitted_mismatched_deregister_skill_keeps_victim_intents(self):
+        # the bus client twins ovos.skill.deregister into the legacy
+        # detach_skill topic with the ORIGINAL context, so the legacy
+        # handler must resolve skill_id from context too: an attacker
+        # naming the victim in the payload leaves the victim's roster intact
+        svc = self.get_service()
+        self._register_victim(svc)
+        svc.bus.emit(Message(self.SpecMessage.SKILL_DEREGISTER.value,
+                             {"skill_id": "victim.skill"},
+                             {"skill_id": "attacker.skill"}))
+        self.assertEqual(list(svc.containers["en-US"].intent_samples),
+                         ["victim.skill:play_music"])
+        self.assertIn("victim.skill:play_music", svc.registered_intents)
+        intent = svc.calc_intent("play jazz", "en-US")
+        self.assertEqual(intent.name, "victim.skill:play_music")
+
+    def test_legacy_detach_skill_mismatched_context_keeps_victim_intents(self):
+        svc = self.get_service()
+        self._register_victim(svc)
+        svc.bus.emit(Message("detach_skill", {"skill_id": "victim.skill"},
+                             {"skill_id": "attacker.skill"}))
+        self.assertEqual(list(svc.containers["en-US"].intent_samples),
+                         ["victim.skill:play_music"])
+
+    def test_legacy_detach_skill_matching_context_removes_intents(self):
+        svc = self.get_service()
+        self._register_victim(svc)
+        svc.bus.emit(Message("detach_skill", {"skill_id": "victim.skill"},
+                             {"skill_id": "victim.skill"}))
+        self.assertEqual(list(svc.containers["en-US"].intent_samples), [])
+        self.assertNotIn("victim.skill:play_music", svc.registered_intents)
+
+    def test_legacy_detach_skill_without_context_keeps_legacy_behaviour(self):
+        svc = self.get_service()
+        self._register_victim(svc)
+        svc.bus.emit(Message("detach_skill", {"skill_id": "victim.skill"}))
+        self.assertEqual(list(svc.containers["en-US"].intent_samples), [])
+
+    def test_detach_intent_forged_prefix_is_rejected(self):
+        # legacy detach_intent carries the owner only as the intent_name
+        # prefix; a producer under another context must not detach it
+        svc = self.get_service()
+        self._register_victim(svc)
+        svc.bus.emit(Message("detach_intent",
+                             {"intent_name": "victim.skill:play_music"},
+                             {"skill_id": "attacker.skill"}))
+        self.assertEqual(list(svc.containers["en-US"].intent_samples),
+                         ["victim.skill:play_music"])
+        self.assertIn("victim.skill:play_music", svc.registered_intents)
+        intent = svc.calc_intent("play jazz", "en-US")
+        self.assertEqual(intent.name, "victim.skill:play_music")
+
+    def test_detach_intent_matching_prefix_detaches(self):
+        svc = self.get_service()
+        self._register_victim(svc)
+        svc.bus.emit(Message("detach_intent",
+                             {"intent_name": "victim.skill:play_music"},
+                             {"skill_id": "victim.skill"}))
+        self.assertEqual(list(svc.containers["en-US"].intent_samples), [])
+        self.assertNotIn("victim.skill:play_music", svc.registered_intents)
+
+    def test_detach_intent_without_context_keeps_legacy_behaviour(self):
+        svc = self.get_service()
+        self._register_victim(svc)
+        svc.bus.emit(Message("detach_intent",
+                             {"intent_name": "victim.skill:play_music"}))
+        self.assertEqual(list(svc.containers["en-US"].intent_samples), [])
+
+    def test_detach_intent_bare_name_with_context_keeps_legacy_behaviour(self):
+        # an unnamespaced legacy name has no owner prefix to compare, so the
+        # guard does not apply (parity with ovos_adapt and ovos_padatious)
+        svc = self.get_service()
+        svc.bus.emit(Message("padatious:register_intent", {
+            "name": "on", "samples": ["turn on the {thing}"],
+            "lang": "en-US"}, {"skill_id": "victim.skill"}))
+        self.assertIn("on", svc.containers["en-US"].intent_samples)
+        svc.bus.emit(Message("detach_intent", {"intent_name": "on"},
+                             {"skill_id": "victim.skill"}))
+        self.assertNotIn("on", svc.containers["en-US"].intent_samples)
+
+    def test_detach_intent_empty_context_skill_id_keeps_legacy_behaviour(self):
+        svc = self.get_service()
+        self._register_victim(svc)
+        svc.bus.emit(Message("detach_intent",
+                             {"intent_name": "victim.skill:play_music"},
+                             {"skill_id": ""}))
+        self.assertEqual(list(svc.containers["en-US"].intent_samples), [])
+
 
 class ContextGatingTest(unittest.TestCase):
     """OVOS-CONTEXT-1 §6/§6.1 requires_context / excludes_context gating."""