Cosmetics

Author: mariadb-AlexanderPresniakov

cmapi/cmapi_server/managers/host_identity.py

@@ -18,9 +18,69 @@
 logger = logging.getLogger(__name__)
 
 
-@lru_cache(maxsize=1)  # singleton
-def get_host_address_manager() -> 'HostAddressManager':
-    return HostAddressManager()
+@dataclass
+class HostIdentity:
+    """Host's network identity, IP addrs and hostnames that are visible from other hosts"""
+    input: str
+    # The first, most important IP addr and host name (ordering is done by policy)
+    primary_ip: str
+    primary_name: Optional[str]  # Name can be missing (but policy can make it required)
+    ips: list[str]  # All IP addrs
+    names: list[str]  # All host names (only those that are visible to other hosts)
+    unique_key: str  # unique id of this host, will be used later for aliases
+    observed_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
+
+    @staticmethod
+    def from_policy(input: str, policy: 'ResolutionPolicy', ips: Sequence[IPAddress], names: Sequence[str]) -> 'HostIdentity':
+        """Construct a HostIdentity from filtered addresses and names using policy ordering."""
+        if not ips:
+            raise ResolutionPolicyViolationError('All resolved addresses were rejected by policy.')
+
+        ordered_ips = policy.order_addresses(ips)
+        # Calculate host unique key from its addresses (we prefer globals as more stable ones)
+        globals_ordered = [ip for ip in ordered_ips if ip.is_global]
+        privates_ordered = [ip for ip in ordered_ips if ip.is_private]
+        to_hash = globals_ordered if globals_ordered else privates_ordered
+        hasher = hashlib.sha256()
+        for ip in to_hash:
+            hasher.update(str(ip).encode('utf-8'))
+        unique_key = hasher.hexdigest()
+
+        return HostIdentity(
+            input=input,
+            ips=[str(ip) for ip in ordered_ips],
+            names=list(names),
+            primary_ip=str(ordered_ips[0]),
+            primary_name=names[0] if names else None,
+            unique_key=unique_key,
+        )
+
+    @property
+    def effective_hostname(self) -> str:
+        """Hostname or primary IP if hostname is not set
+
+        If policy requires host to have a hostname and it doesn't, we won't even get here
+        """
+        return self.primary_name or self.primary_ip
+
+    def __repr__(self) -> str:
+        parts: list[str] = [
+            f'input={self.input!r}',
+            f'primary_ip={self.primary_ip!r}',
+        ]
+        if self.primary_name is not None:
+            parts.append(f'primary_name={self.primary_name!r}')
+        # Don't show addrs and hostnames if there is only one addr or hostname
+        if self.ips != [self.primary_ip]:
+            parts.append(f'ips={self.ips!r}')
+        if self.names != [self.primary_name]:
+            parts.append(f'names={self.names!r}')
+        parts.append(f'unique_key={self.unique_key[:4]}')
+        parts.append(f'observed_at={self.observed_at.replace(microsecond=0).isoformat()!r}')
+        return 'HostIdentity(' + ', '.join(parts) + ')'
+
+    def __str__(self) -> str:
+        return repr(self)
 
 
 @dataclass(frozen=True)
@@ -44,9 +104,9 @@ def filter_addresses(self, addrs: Iterable[str]) -> list[IPAddress]:
             list(addrs), self.allow_private_ips, self.allow_ipv6
         )
         for addr in addrs:
-            ip = ip_or_none(addr)
+            ip = _ip_or_none(addr)
             if ip is None:
-                logger.debug('Skip %s: not a valid IP literal', addr)
+                logger.error('Skip %s: not a valid IP literal', addr)
                 continue
 
             # Fixed rejections first
@@ -91,71 +151,6 @@ def key(ip: IPAddress) -> tuple[int, int, int, int, int, int, int, int]:
         return sorted(ips, key=key)
 
 
-@dataclass
-class HostIdentity:
-    """Host's network identity, IP addrs and hostnames that are visible from other hosts"""
-    input: str
-    # The first, most important IP addr and host name (ordering is done by policy)
-    primary_ip: str
-    primary_name: Optional[str]  # Name can be missing (but policy can make it required)
-    ips: list[str]  # All IP addrs
-    names: list[str]  # All host names (only those that are visible to other hosts)
-    unique_key: str  # unique id of this host, will be used later for aliases
-    observed_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
-
-    @staticmethod
-    def from_policy(input: str, policy: ResolutionPolicy, ips: Sequence[IPAddress], names: Sequence[str]) -> 'HostIdentity':
-        """Construct a HostIdentity from filtered addresses and names using policy ordering."""
-        if not ips:
-            raise ResolutionPolicyViolationError('All resolved addresses were rejected by policy.')
-
-        ordered_ips = policy.order_addresses(ips)
-        # Calculate host unique key from its addresses (we prefer globals as more stable ones)
-        globals_ordered = [ip for ip in ordered_ips if ip.is_global]
-        privates_ordered = [ip for ip in ordered_ips if ip.is_private]
-        to_hash = globals_ordered if globals_ordered else privates_ordered
-        hasher = hashlib.sha256()
-        for ip in to_hash:
-            hasher.update(str(ip).encode('utf-8'))
-        unique_key = hasher.hexdigest()
-
-        return HostIdentity(
-            input=input,
-            ips=[str(ip) for ip in ordered_ips],
-            names=list(names),
-            primary_ip=str(ordered_ips[0]),
-            primary_name=names[0] if names else None,
-            unique_key=unique_key,
-        )
-
-    @property
-    def effective_hostname(self) -> str:
-        """Hostname or primary IP if hostname is not set
-
-        If policy requires host to have a hostname and it doesn't, we won't even get here
-        """
-        return self.primary_name or self.primary_ip
-
-    def __repr__(self) -> str:
-        parts: list[str] = [
-            f'input={self.input!r}',
-            f'primary_ip={self.primary_ip!r}',
-        ]
-        if self.primary_name is not None:
-            parts.append(f'primary_name={self.primary_name!r}')
-        # Don't show addrs and hostnames if there is only one addr or hostname
-        if self.ips != [self.primary_ip]:
-            parts.append(f'ips={self.ips!r}')
-        if self.names != [self.primary_name]:
-            parts.append(f'names={self.names!r}')
-        parts.append(f'unique_key={self.unique_key[:4]}')
-        parts.append(f'observed_at={self.observed_at.replace(microsecond=0).isoformat()!r}')
-        return 'HostIdentity(' + ', '.join(parts) + ')'
-
-    def __str__(self) -> str:
-        return repr(self)
-
-
 class HostAddressManager:
     """Calculates HostIdentity from passed hostname or IP address."""
 
@@ -170,7 +165,7 @@ def get_identity(self, target: str) -> HostIdentity:
 
         target = target.strip()
 
-        ip = ip_or_none(target)
+        ip = _ip_or_none(target)
         if ip is not None:
             identity = self._get_identity_from_ip(ip, target)
         else:
@@ -211,7 +206,7 @@ def check_hostname_rev_lookup(self, hostname: str) -> tuple[HostIdentity, bool]:
             return identity, False
 
         for ip_text in identity.ips:
-            ip = ip_or_none(ip_text)
+            ip = _ip_or_none(ip_text)
             if ip is None:
                 logger.error('Invalid IP address: %s', ip_text)
                 continue
@@ -229,10 +224,6 @@ def check_hostname_rev_lookup(self, hostname: str) -> tuple[HostIdentity, bool]:
         logger.warning('Roundtrip check failed for %s', hostname)
         return identity, False
 
-    @property
-    def policy(self) -> ResolutionPolicy:
-        return self._policy
-
     def _get_identity_from_ip(self, ip: IPAddress, original_input: str) -> HostIdentity:
         if not self._policy.filter_addresses([str(ip)]):
             raise ResolutionPolicyViolationError('Input IP address was rejected by policy')
@@ -246,9 +237,10 @@ def _get_identity_from_ip(self, ip: IPAddress, original_input: str) -> HostIdent
 
     def _get_identity_from_hostname(self, hostname: str) -> HostIdentity:
         normalized = hostname.strip().lower()
-        if not _is_fqdn(normalized):
+        if _is_fqdn(normalized):
+            return self._get_identity_from_fqdn(normalized)
+        else:
             return self._get_identity_from_non_fqdn(hostname)
-        return self._get_identity_from_fqdn(normalized)
 
     def _get_identity_from_fqdn(self, fqdn: str) -> HostIdentity:
         # Get IPs from hostname (via DNS), filter them, then get names from each IP
@@ -296,7 +288,7 @@ def _get_identity_from_non_fqdn(self, hostname: str) -> HostIdentity:
         # Collect names of IPs
         names: set[str] = set()
         for ip_text in list(candidate_ips):
-            ip = ip_or_none(ip_text)
+            ip = _ip_or_none(ip_text)
             if ip is None:
                 logger.error('Invalid IP address: %s', ip_text)
                 continue
@@ -337,7 +329,7 @@ def _resolve_dns(self, hostname: str) -> list[IPAddress]:
 
         addrs: list[IPAddress] = []
         for ip_text in ipv4_texts + ipv6_texts:
-            ip = ip_or_none(ip_text)
+            ip = _ip_or_none(ip_text)
             if ip is None:
                 logger.error('DNS returned invalid IP address %s for host name %s, skipping', ip_text, hostname)
                 continue
@@ -393,21 +385,27 @@ def _dns_reverse(self, ip_text: str) -> list[str]:
     def _contains_private(self, addrs: list[str]) -> bool:
         """Return True if any resolvable address string is a private IP."""
         for addr in addrs:
-            ip = ip_or_none(addr)
+            ip = _ip_or_none(addr)
             if ip is None:
                 continue
             if ip.is_private:
                 return True
         return False
 
-def ip_or_none(val: str) -> Optional[IPAddress]:
+
+@lru_cache(maxsize=1)  # singleton
+def get_host_address_manager() -> 'HostAddressManager':
+    return HostAddressManager()
+
+
+def _ip_or_none(val: str) -> Optional[IPAddress]:
     try:
         return ipaddress.ip_address(val)
     except ValueError:
         return None
 
-def is_ip_address(val: str) -> bool:
-    return ip_or_none(val) is not None
+def _is_ip_address(val: str) -> bool:
+    return _ip_or_none(val) is not None
 
 def _is_fqdn(name: str) -> bool:
     """Return True if the string is a valid FQDN (lower-cased, no trailing dot).