Skip to content

decapsulate() uses string rindex — produces incorrect results for repeated protocols #109

Description

@sumanjeet0012

Multiaddr.decapsulate() converts both multiaddrs to strings and uses str.rindex() to find the decapsulation point. This produces incorrect results when the same protocol appears multiple times in the address, or when the decapsulation target is a substring of a protocol value.

Problem

In multiaddr/multiaddr.py:

def decapsulate(self, addr):
    addr_str = str(addr)
    s = str(self)
    i = s.rindex(addr_str)        # ← String search, not component-level
    if i < 0:
        raise ValueError(...)
    return Multiaddr(s[:i])

Case 1 — Repeated protocol:

ma = Multiaddr("/ip4/1.2.3.4/tcp/80/ip4/5.6.7.8/tcp/443")
result = ma.decapsulate("/ip4/5.6.7.8")
# Expected: /ip4/1.2.3.4/tcp/80
# Actual:   /ip4/1.2.3.4/tcp/80  ✅ (happens to work because rindex finds last)

Case 2 — Substring collision:

ma = Multiaddr("/dns4/example.com/tcp/80")
result = ma.decapsulate("/tcp/8")
# rindex finds "tcp/8" inside "tcp/80" → incorrect split point

Case 3 — Value contains protocol name:

ma = Multiaddr("/dns4/tcp.example.com/tcp/80")
result = ma.decapsulate("/tcp/80")
# rindex might find "tcp" inside "tcp.example.com" → incorrect

Go's implementation does component-by-component comparison:

func (m Multiaddr) Decapsulate(rightPartsAny Multiaddrer) Multiaddr {
    rightParts := rightPartsAny.Multiaddr()
    leftParts := m
    lastIndex := -1
    for i := range leftParts {
        foundMatch := false
        for j, rightC := range rightParts {
            if len(leftParts) <= i+j { break }
            foundMatch = rightC.Equal(&leftParts[i+j])
            if !foundMatch { break }
        }
        if foundMatch { lastIndex = i }
    }
    // ...
}

Proposed Solution

Rewrite decapsulate() to compare at the component level using bytes_iter():

def decapsulate(self, addr):
    other = Multiaddr(addr) if not isinstance(addr, Multiaddr) else addr
    other_components = list(bytes_iter(other.to_bytes()))

    self_components = list(bytes_iter(self._bytes))
    last_match_end = -1

    for i in range(len(self_components)):
        match = True
        for j, (offset, proto, codec, value) in enumerate(other_components):
            if i + j >= len(self_components):
                match = False
                break
            s_offset, s_proto, s_codec, s_value = self_components[i + j]
            if s_proto != proto or s_value != value:
                match = False
                break
        if match:
            last_match_end = self_components[i][0]  # byte offset

    if last_match_end < 0:
        raise ValueError(f"Address {self} does not contain subaddress: {addr}")
    if last_match_end == 0:
        return Multiaddr("")
    return Multiaddr(self._bytes[:last_match_end])

Related

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Type

    No type

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions