NameEvent support also host identification by tag

tcsfw/address.py

@@ -202,7 +202,7 @@ def get_tag(cls, addresses: Iterable[AnyAddress]) -> Optional[EntityTag]:
 
     @classmethod
     def parse_address(cls, address: str) -> AnyAddress:
-        """Parse any address type from string, type given as 'type/address'"""
+        """Parse any address type from string, type given as 'type|address'"""
         v, _, t = address.rpartition("|")
         if v == "":
             t, v = "ip", t  # default is IP

tcsfw/inspector.py

@@ -3,7 +3,7 @@
 import logging
 from typing import Dict, Optional, Set
 
-from tcsfw.address import DNSName, AnyAddress
+from tcsfw.address import AnyAddress
 from tcsfw.basics import ExternalActivity, Status
 from tcsfw.entity import Entity
 from tcsfw.event_interface import EventInterface, PropertyAddressEvent, PropertyEvent
@@ -151,7 +151,7 @@ def name(self, event: NameEvent) -> Optional[Host]:
         address = event.address
         if event.service and event.service.captive_portal and event.address in event.service.parent.addresses:
             address = None  # it is just redirecting to itself
-        name = DNSName(event.name)
+        name = event.tag or event.name
         h, changes = self.system.learn_named_address(name, address)
         if h not in self.known_entities:
             # new host

tcsfw/mitm_log_reader.py

@@ -3,7 +3,7 @@
 from io import BytesIO, TextIOWrapper
 import re
 
-from tcsfw.address import HWAddresses
+from tcsfw.address import DNSName, HWAddresses
 from tcsfw.event_interface import EventInterface
 from tcsfw.model import IoTSystem
 from tcsfw.property import Properties
@@ -62,7 +62,7 @@ def process_file(self, data: BytesIO, file_name: str, interface: EventInterface,
                 flow.evidence = evidence
                 if d:
                     # learn SNI, no peers in event, the connection will be UNEXPECTED if it is not expected
-                    name = NameEvent(evidence, None, d, flow.target[1])
+                    name = NameEvent(evidence, None, name=DNSName(d), address=flow.target[1])
                     if name not in names:
                         interface.name(name)
                         names.add(name)

tcsfw/model.py

@@ -3,7 +3,7 @@
 import ipaddress
 import itertools
 import re
-from typing import List, Set, Optional, Tuple, TypeVar, Callable, Dict, Any, Self, Iterable, Iterator
+from typing import List, Set, Optional, Tuple, TypeVar, Callable, Dict, Any, Self, Iterable, Iterator, Union
 from urllib.parse import urlparse
 
 from tcsfw.address import AnyAddress, Addresses, EndpointAddress, EntityTag, Protocol, IPAddress, HWAddress, DNSName
@@ -484,27 +484,28 @@ def is_external(self, address: AnyAddress) -> bool:
                 return False
         return True
 
-    def learn_named_address(self, name: DNSName, address: Optional[AnyAddress]) -> Tuple[Host, bool]:
-        """Learn DNS named addresses, return named host and if any changes"""
+    def learn_named_address(self, name: Union[DNSName, EntityTag], address: Optional[AnyAddress]) -> Tuple[Host, bool]:
+        """Learn addresses for host, return the named host and if any changes"""
         # pylint: disable=too-many-return-statements
 
-        # check for reverse DNS
-        if name.name.endswith(".arpa") and len(name.name) > 5:
-            # reverse DNS from IP addresss to name
-            nn = name.name[:-5]
-            if nn.endswith(".in-addr") and len(nn) > 8:
-                address = IPAddress.new(nn[:-8])
-            elif nn.endswith(".ip6") and len(nn) > 4:
-                nn = nn[:-4].replace(".", "")[::-1]
-                nn = ":".join(re.findall("....", nn))
-                address = IPAddress.new(nn)
-            else:
-                # E.g. _dns.resolver.arpa - leave as name!
-                address = None
-            if address:
-                add = self.get_endpoint(address)
-                assert isinstance(add, Host)
-                return add, False  # Did not add name to host (why?)
+        if isinstance(name, DNSName):
+            # check for reverse DNS
+            if name.name.endswith(".arpa") and len(name.name) > 5:
+                # reverse DNS from IP addresss to name
+                nn = name.name[:-5]
+                if nn.endswith(".in-addr") and len(nn) > 8:
+                    address = IPAddress.new(nn[:-8])
+                elif nn.endswith(".ip6") and len(nn) > 4:
+                    nn = nn[:-4].replace(".", "")[::-1]
+                    nn = ":".join(re.findall("....", nn))
+                    address = IPAddress.new(nn)
+                else:
+                    # E.g. _dns.resolver.arpa - leave as name!
+                    address = None
+                if address:
+                    add = self.get_endpoint(address)
+                    assert isinstance(add, Host)
+                    return add, False  # Did not add name to host (why?)
 
         # find relevant hosts
         named = None
@@ -533,6 +534,8 @@ def learn_named_address(self, name: DNSName, address: Optional[AnyAddress]) -> T
             return add, True
 
         if named is None:
+            if isinstance(name, EntityTag):
+                return None, False  # do not create hosts for unknown tags
             named = self.get_endpoint(name)
 
         if not add:

tcsfw/pcap_reader.py

@@ -16,7 +16,7 @@
 from framing.frames import Frames
 from framing.raw_data import Raw, RawData
 
-from tcsfw.address import HWAddress, Protocol, IPAddress
+from tcsfw.address import DNSName, HWAddress, Protocol, IPAddress
 from tcsfw.event_interface import EventInterface
 from tcsfw.model import Connection, IoTSystem
 from tcsfw.services import NameEvent, DNSService
@@ -169,14 +169,14 @@ def _dns_message(self, peers: Tuple[IPAddress], udp: UDP, connection: Connection
             name = dns_frames.DNSName.string(rd, dns_frames.DNSQuestion.QNAME)
             if name not in self.dns_names:
                 self.dns_names[name] = None
-                events.append(NameEvent(evidence, service, name, peers=peers))
+                events.append(NameEvent(evidence, service, name=DNSName(name), peers=peers))
 
         def learn_name(name: str, ip: IPAddress):
             old = self.dns_names.get(ip)
             if old == name:
                 return
             self.dns_names[ip] = name
-            n = NameEvent(evidence, service, name, IPAddress(ip), peers=peers)
+            n = NameEvent(evidence, service, name=DNSName(name), address=IPAddress(ip), peers=peers)
             events.append(n)
 
         rd_frames = []

tcsfw/services.py

@@ -2,7 +2,7 @@
 
 from typing import Any, Callable, Dict, List, Set, Optional
 
-from tcsfw.address import EndpointAddress, Protocol, IPAddress
+from tcsfw.address import DNSName, EndpointAddress, EntityTag, Protocol, IPAddress
 from tcsfw.basics import ConnectionType, HostType
 from tcsfw.model import Service, NetworkNode, Connection, Host, Addressable
 from tcsfw.traffic import IPFlow, Flow, Event, Evidence
@@ -42,22 +42,27 @@ def __init__(self, parent: Addressable, name="DNS"):
 
 
 class NameEvent(Event):
-    """DNS name event"""
-    def __init__(self, evidence: Evidence, service: Optional[DNSService], name: str,
-                 address: Optional[IPAddress] = None, peers: List[NetworkNode] = None):
+    """Name or tag and address event"""
+    def __init__(self, evidence: Evidence, service: Optional[DNSService], name: Optional[DNSName] = None,
+                 tag: Optional[EntityTag] = None, address: Optional[IPAddress] = None,
+                 peers: List[NetworkNode] = None):
         super().__init__(evidence)
+        assert bool(name) != bool(tag), "Name or tag must be set"
         self.service = service
         self.name = name
+        self.tag = tag
         self.address = address
         self.peers = [] if peers is None else peers  # The communicating peers
 
     def get_value_string(self) -> str:
-        return f"{self.name}={self.address}" if self.address else self.name
+        return f"{self.name or self.tag}={self.address}" if self.address else str(self.name or self.tag)
 
     def get_data_json(self, id_resolver: Callable[[Any], Any]) -> Dict:
-        r = {
-            "name": self.name,
-        }
+        r = {}
+        if self.name:
+            r["name"] = self.name.name
+        if self.tag:
+            r["tag"] = self.tag.tag
         if self.service:
             r["service"] = id_resolver(self.service)
         if self.address:
@@ -69,21 +74,22 @@ def get_data_json(self, id_resolver: Callable[[Any], Any]) -> Dict:
     @classmethod
     def decode_data_json(cls, evidence: Evidence, data: Dict, entity_resolver: Callable[[Any], Any]):
         """Decode event from JSON"""
-        name = data["name"]
+        name = DNSName(data.get("name")) if "name" in data else None
+        tag = EntityTag(data.get("tag")) if "tag" in data else None
         service = entity_resolver(data.get("service")) if "service" in data else None
         assert service is None or isinstance(service, DNSService), f"Bad service {service.__class__.__name__}"
         address = IPAddress.new(data.get("address")) if "address" in data else None
         peers = [entity_resolver(p) for p in data.get("peers", [])]
         assert all(p for p in peers)
-        return NameEvent(evidence, service, name, address, peers)
+        return NameEvent(evidence, service, name, tag, address, peers)
 
     def __eq__(self, other):
         if not isinstance(other, NameEvent):
             return False
         return self.service == other.service and self.name == other.name and self.address == other.address
 
     def __hash__(self):
-        return self.name.__hash__() ^ (self.address.__hash__() if self.address else 0)
+        return hash(self.name) ^ hash(self.tag) ^ hash(self.address)
 
     def __repr__(self):
-        return f"{self.address}" + (f" '{self.name}'" if self.name else "")
+        return self.get_value_string()

tcsfw/setup_reader.py

@@ -2,10 +2,12 @@
 import csv
 
 from io import BytesIO, TextIOWrapper
+from tcsfw.address import DNSName, EntityTag
 from tcsfw.event_interface import EventInterface
 from tcsfw.model import IoTSystem
+from tcsfw.services import NameEvent
 from tcsfw.tools import ToolAdapter
-from tcsfw.traffic import EvidenceSource
+from tcsfw.traffic import Evidence, EvidenceSource
 
 
 class SetupCSVReader(ToolAdapter):
@@ -19,6 +21,7 @@ def process_file(self, data: BytesIO, file_name: str, interface: EventInterface,
         columns = {}
         host_i = -1
         address_i = -1
+        ev = Evidence(source)
         for row in reader:
             if not columns:
                 columns = {c: i for i, c in enumerate(row)}
@@ -32,5 +35,8 @@ def process_file(self, data: BytesIO, file_name: str, interface: EventInterface,
             ads = row[address_i].strip().split(", \t\n\r")
             if not host_tag or not ads:
                 continue
-            # FIXME: Does nothing now!
+            for a in ads:
+                address = DNSName.name_or_ip(a)
+                event = NameEvent(ev, None, tag=EntityTag(host_tag), address=address)
+                interface.name(event)
         return True

tests/samples/setup-doc/setup.csv

@@ -1,2 +1,3 @@
 Host,Address
+Device,example.com
 Device,192.168.2.5

tests/test_event.py

@@ -1,4 +1,4 @@
-from tcsfw.address import Addresses, EndpointAddress, Protocol
+from tcsfw.address import Addresses, DNSName, EndpointAddress, Protocol
 from tcsfw.verdict import Verdict
 from tcsfw.builder_backend import SystemBackend
 from tcsfw.event_interface import PropertyAddressEvent, PropertyEvent
@@ -74,7 +74,7 @@ def test_name_event():
     }
     ent_reverse = {v: k for k, v in entities.items()}
 
-    p = NameEvent(evi, service.entity, "www.example.com")
+    p = NameEvent(evi, service.entity, name=DNSName("www.example.com"))
     js = p.get_data_json(entities.get)
     assert js == {
         'service': 12,
@@ -83,7 +83,7 @@ def test_name_event():
 
     p2 = NameEvent.decode_data_json(evi, js, ent_reverse.get)
     assert p2.service == service.entity
-    assert p2.name == "www.example.com"
+    assert p2.name == DNSName("www.example.com")
     assert p == p2
 
 

tests/test_inspector.py

@@ -2,7 +2,7 @@
 from tcsfw.builder_backend import SystemBackend
 from tcsfw.services import NameEvent
 import test_model
-from tcsfw.address import EndpointAddress, Protocol, IPAddress
+from tcsfw.address import DNSName, EndpointAddress, Protocol, IPAddress
 from tcsfw.inspector import Inspector
 from tcsfw.main import DHCP, DNS, UDP, TCP
 from tcsfw.traffic import NO_EVIDENCE, IPFlow, Evidence, EvidenceSource, ServiceScan, HostScan
@@ -196,7 +196,7 @@ def test_learn_dns_name():
     i.connection(flow_0)
 
     # event about DNS naming
-    ev = NameEvent(NO_EVIDENCE, dns.entity, "Aname.org", IPAddress.new("12.0.0.2"))
+    ev = NameEvent(NO_EVIDENCE, dns.entity, name=DNSName("Aname.org"), address=IPAddress.new("12.0.0.2"))
     i.name(ev)
 
     flow = IPFlow.UDP("1:0:0:0:0:1", "22.0.0.1", 1100) >> ("1:0:0:0:0:2", "12.0.0.2", 1234)
@@ -220,7 +220,7 @@ def test_learn_dns_name_expected_connection():
     i.connection(flow_0)
 
     # event about DNS naming
-    ev = NameEvent(NO_EVIDENCE, dns.entity, "Aname.org", IPAddress.new("12.0.0.2"))
+    ev = NameEvent(NO_EVIDENCE, dns.entity, name=DNSName("Aname.org"), address=IPAddress.new("12.0.0.2"))
     i.name(ev)
 
     flow = IPFlow.UDP("1:0:0:0:0:1", "192.168.0.2", 1100) >> ("1:0:0:0:0:2", "12.0.0.2", 1234)

tests/test_setup_reader.py

@@ -2,6 +2,7 @@
 
 import pathlib
 
+from tcsfw.address import DNSName, EntityTag, IPAddress
 from tcsfw.batch_import import BatchImporter
 from tests.test_model import Setup
 
@@ -17,4 +18,6 @@ def __init__(self):
 def test_setup_csv():
     su = Setup_1()
     BatchImporter(su.get_inspector()).import_batch(pathlib.Path("tests/samples/setup-doc"))
+    assert su.device1.entity.addresses == {EntityTag("Device"), DNSName("example.com"), IPAddress.new("192.168.2.5")}
+    assert su.device2.entity.addresses == {EntityTag("Device_2")}