What Is a Node in Networking? Definition, Types, and How They Work

from dataclasses import dataclass, field
from enum import Enum
from typing import List, Optional, Dict
from io.thecodeforge.network.models import NetworkAddress


class NodeType(Enum):
    ENDPOINT = "endpoint"
    ROUTER = "router"
    SWITCH = "switch"
    FIREWALL = "firewall"
    LOAD_BALANCER = "load_balancer"
    SERVER = "server"
    IOT_DEVICE = "iot_device"
    VIRTUAL = "virtual"


class NodeRole(Enum):
    BACKBONE = "backbone"
    DISTRIBUTION = "distribution"
    ACCESS = "access"
    EDGE = "edge"
    ENDPOINT = "endpoint"


@dataclass
class NetworkNode:
    """
    Represents a network node with addressing, role classification,
    and health monitoring attributes.
    """
    
    node_id: str
    hostname: str
    node_type: NodeType
    role: NodeRole
    ip_addresses: List[str] = field(default_factory=list)
    mac_addresses: List[str] = field(default_factory=list)
    interfaces: List[str] = field(default_factory=list)
    is_reachable: bool = True
    latency_ms: float = 0.0
    packet_loss_percent: float = 0.0
    uptime_seconds: float = 0.0
    
    @property
    def is_critical(self) -> bool:
        return self.role in (NodeRole.BACKBONE, NodeRole.DISTRIBUTION)
    
    @property
    def health_score(self) -> float:
        """
        Calculate node health score from 0.0 (down) to 1.0 (healthy).
        """
        if not self.is_reachable:
            return 0.0
        
        latency_penalty = min(self.latency_ms / 100.0, 0.3)
        loss_penalty = min(self.packet_loss_percent / 10.0, 0.5)
        
        return max(0.0, 1.0 - latency_penalty - loss_penalty)


class NetworkTopology:
    """
    Manages a collection of network nodes and their interconnections.
    """
    
    def __init__(self):
        self.nodes: Dict[str, NetworkNode] = {}
        self.adjacency: Dict[str, List[str]] = {}
    
    def add_node(self, node: NetworkNode) -> None:
        self.nodes[node.node_id] = node
        if node.node_id not in self.adjacency:
            self.adjacency[node.node_id] = []
    
    def add_link(self, node_a: str, node_b: str) -> None:
        if node_a not in self.adjacency:
            self.adjacency[node_a] = []
        if node_b not in self.adjacency:
            self.adjacency[node_b] = []
        
        if node_b not in self.adjacency[node_a]:
            self.adjacency[node_a].append(node_b)
        if node_a not in self.adjacency[node_b]:
            self.adjacency[node_b].append(node_a)
    
    def find_critical_nodes(self) -> List[NetworkNode]:
        """
        Identify nodes whose failure would partition the network.
        These are articulation points in the topology graph.
        """
        critical = []
        
        for node_id, node in self.nodes.items():
            if node.is_critical:
                critical.append(node)
            elif len(self.adjacency.get(node_id, [])) == 1:
                critical.append(node)
        
        return critical
    
    def classify_nodes(self) -> Dict[NodeType, List[NetworkNode]]:
        """
        Group nodes by type for inventory and monitoring.
        """
        classified = {}
        for node in self.nodes.values():
            if node.node_type not in classified:
                classified[node.node_type] = []
            classified[node.node_type].append(node)
        return classified


# Example topology
topology = NetworkTopology()

topology.add_node(NetworkNode(
    node_id="core-sw-01", hostname="core-switch-01",
    node_type=NodeType.SWITCH, role=NodeRole.BACKBONE,
    ip_addresses=["10.0.0.1"], interfaces=["eth0", "eth1", "eth2"]
))

topology.add_node(NetworkNode(
    node_id="web-srv-01", hostname="web-server-01",
    node_type=NodeType.SERVER, role=NodeRole.ENDPOINT,
    ip_addresses=["10.0.1.10"], interfaces=["eth0"]
))

topology.add_link("core-sw-01", "web-srv-01")

for node in topology.find_critical_nodes():
    print(f"Critical: {node.hostname} ({node.role.value})")

from dataclasses import dataclass
from typing import List, Dict, Optional
from io.thecodeforge.network.node_classifier import NodeType, NodeRole, NetworkNode


@dataclass
class NodeTypeCapabilities:
    node_type: str
    osi_layer: int
    forwarding_method: str
    address_type: str
    typical_redundancy: str
    failure_blast_radius: str


class NodeTypeRegistry:
    """
    Registry of network node types with their capabilities
    and operational characteristics.
    """
    
    TYPE_DEFINITIONS = {
        NodeType.ROUTER: NodeTypeCapabilities(
            node_type="Router",
            osi_layer=3,
            forwarding_method="IP routing table lookup",
            address_type="IP address",
            typical_redundancy="VRRP/HSRP or ECMP",
            failure_blast_radius="All traffic between connected networks"
        ),
        NodeType.SWITCH: NodeTypeCapabilities(
            node_type="Switch",
            osi_layer=2,
            forwarding_method="MAC address table lookup",
            address_type="MAC address",
            typical_redundancy="STP/RSTP or MLAG",
            failure_blast_radius="All devices on connected segments"
        ),
        NodeType.FIREWALL: NodeTypeCapabilities(
            node_type="Firewall",
            osi_layer=3,
            forwarding_method="Stateful packet inspection",
            address_type="IP address",
            typical_redundancy="Active-passive HA pair",
            failure_blast_radius="All traffic crossing security boundary"
        ),
        NodeType.LOAD_BALANCER: NodeTypeCapabilities(
            node_type="Load Balancer",
            osi_layer=4,
            forwarding_method="Connection distribution algorithm",
            address_type="Virtual IP (VIP)",
            typical_redundancy="Active-active with health checks",
            failure_blast_radius="All services behind the VIP"
        ),
        NodeType.SERVER: NodeTypeCapabilities(
            node_type="Server",
            osi_layer=7,
            forwarding_method="Application-level processing",
            address_type="IP address",
            typical_redundancy="Horizontal scaling with load balancer",
            failure_blast_radius="Services hosted on this server"
        ),
        NodeType.ENDPOINT: NodeTypeCapabilities(
            node_type="Endpoint",
            osi_layer=7,
            forwarding_method="None — source or destination only",
            address_type="IP and MAC address",
            typical_redundancy="None — individual device",
            failure_blast_radius="Single user or service"
        )
    }
    
    @staticmethod
    def get_capabilities(node_type: NodeType) -> Optional[NodeTypeCapabilities]:
        return NodeTypeRegistry.TYPE_DEFINITIONS.get(node_type)
    
    @staticmethod
    def get_redundancy_requirements(node_type: NodeType) -> str:
        caps = NodeTypeRegistry.get_capabilities(node_type)
        return caps.typical_redundancy if caps else "Unknown"
    
    @staticmethod
    def classify_by_blast_radius(
        nodes: List[NetworkNode]
    ) -> Dict[str, List[NetworkNode]]:
        """
        Group nodes by failure blast radius for risk assessment.
        """
        result = {"high": [], "medium": [], "low": []}
        
        for node in nodes:
            caps = NodeTypeRegistry.get_capabilities(node.node_type)
            if not caps:
                result["medium"].append(node)
                continue
            
            if node.role in (NodeRole.BACKBONE, NodeRole.DISTRIBUTION):
                result["high"].append(node)
            elif node.node_type in (NodeType.FIREWALL, NodeType.LOAD_BALANCER):
                result["high"].append(node)
            elif node.node_type == NodeType.SWITCH:
                result["medium"].append(node)
            else:
                result["low"].append(node)
        
        return result


# Example
for ntype, caps in NodeTypeRegistry.TYPE_DEFINITIONS.items():
    print(f"{caps.node_type}: Layer {caps.osi_layer}, Blast radius: {caps.failure_blast_radius}")

from dataclasses import dataclass
from typing import List, Dict, Optional, Tuple
from enum import Enum


class ProtocolLayer(Enum):
    PHYSICAL = 1
    DATA_LINK = 2
    NETWORK = 3
    TRANSPORT = 4
    SESSION = 5
    PRESENTATION = 6
    APPLICATION = 7


@dataclass
class PacketTrace:
    hop_number: int
    node_hostname: str
    node_ip: str
    ingress_interface: str
    egress_interface: str
    latency_ms: float
    ttl_remaining: int
    action: str


class NodeCommunicationTracer:
    """
    Traces packet flow through network nodes for debugging
    and performance analysis.
    """
    
    @staticmethod
    def trace_route(
        source: str,
        destination: str,
        hops: List[Dict]
    ) -> List[PacketTrace]:
        """
        Simulate a packet trace through network nodes.
        """
        trace = []
        for i, hop in enumerate(hops):
            trace.append(PacketTrace(
                hop_number=i + 1,
                node_hostname=hop["hostname"],
                node_ip=hop["ip"],
                ingress_interface=hop.get("ingress", "N/A"),
                egress_interface=hop.get("egress", "N/A"),
                latency_ms=hop.get("latency_ms", 0.0),
                ttl_remaining=64 - (i + 1),
                action=hop.get("action", "forward")
            ))
        return trace
    
    @staticmethod
    def identify_protocol_layers(node_type: str) -> List[ProtocolLayer]:
        """
        Determine which protocol layers a node type operates on.
        """
        layer_map = {
            "switch": [ProtocolLayer.PHYSICAL, ProtocolLayer.DATA_LINK],
            "router": [ProtocolLayer.PHYSICAL, ProtocolLayer.DATA_LINK, ProtocolLayer.NETWORK],
            "firewall": [ProtocolLayer.PHYSICAL, ProtocolLayer.DATA_LINK, ProtocolLayer.NETWORK, ProtocolLayer.TRANSPORT],
            "load_balancer": [ProtocolLayer.PHYSICAL, ProtocolLayer.DATA_LINK, ProtocolLayer.NETWORK, ProtocolLayer.TRANSPORT, ProtocolLayer.APPLICATION],
            "server": [layer for layer in ProtocolLayer],
            "endpoint": [layer for layer in ProtocolLayer]
        }
        return layer_map.get(node_type.lower(), [ProtocolLayer.PHYSICAL])
    
    @staticmethod
    def resolve_address_at_layer(
        destination: str,
        layer: ProtocolLayer,
        arp_table: Dict[str, str],
        routing_table: List[Dict]
    ) -> Optional[str]:
        """
        Resolve the next-hop address at a specific protocol layer.
        """
        if layer == ProtocolLayer.DATA_LINK:
            return arp_table.get(destination)
        elif layer == ProtocolLayer.NETWORK:
            for route in routing_table:
                if destination.startswith(route["prefix"]):
                    return route["next_hop"]
        return None


# Example trace
tracer = NodeCommunicationTracer()
trace = tracer.trace_route(
    source="10.0.1.10",
    destination="10.0.2.20",
    hops=[
        {"hostname": "access-sw-01", "ip": "10.0.1.1", "latency_ms": 0.2, "action": "forward"},
        {"hostname": "core-rtr-01", "ip": "10.0.0.1", "latency_ms": 0.5, "action": "forward"},
        {"hostname": "dist-sw-01", "ip": "10.0.2.1", "latency_ms": 0.3, "action": "forward"},
        {"hostname": "web-srv-02", "ip": "10.0.2.20", "latency_ms": 0.1, "action": "deliver"}
    ]
)

for hop in trace:
    print(f"Hop {hop.hop_number}: {hop.node_hostname} ({hop.node_ip}) - {hop.latency_ms}ms - TTL:{hop.ttl_remaining}")

from dataclasses import dataclass
from enum import Enum
from typing import List, Dict, Optional
from io.thecodeforge.network.node_classifier import NodeType, NetworkNode


class RedundancyType(Enum):
    ACTIVE_ACTIVE = "active_active"
    ACTIVE_PASSIVE = "active_passive"
    ECMP = "ecmp"
    VRRP = "vrrp"
    MLAG = "mlag"
    ANycast = "anycast"


@dataclass
class RedundancyGroup:
    """
    A group of nodes providing redundant service.
    """
    
    group_id: str
    redundancy_type: RedundancyType
    primary_node: str
    secondary_nodes: List[str]
    virtual_ip: Optional[str] = None
    failover_time_ms: float = 0.0
    state_sync_enabled: bool = False
    
    @property
    def total_nodes(self) -> int:
        return 1 + len(self.secondary_nodes)
    
    @property
    def is_healthy(self) -> bool:
        return self.total_nodes >= 2


class RedundancyPlanner:
    """
    Plans redundancy strategies for network nodes based on
    node type and criticality.
    """
    
    RECOMMENDED_STRATEGIES = {
        NodeType.ROUTER: {
            "primary": RedundancyType.VRRP,
            "alternative": RedundancyType.ECMP,
            "min_nodes": 2,
            "target_failover_ms": 1000,
            "state_sync": False
        },
        NodeType.SWITCH: {
            "primary": RedundancyType.MLAG,
            "alternative": RedundancyType.ACTIVE_ACTIVE,
            "min_nodes": 2,
            "target_failover_ms": 500,
            "state_sync": False
        },
        NodeType.FIREWALL: {
            "primary": RedundancyType.ACTIVE_PASSIVE,
            "alternative": RedundancyType.ACTIVE_ACTIVE,
            "min_nodes": 2,
            "target_failover_ms": 3000,
            "state_sync": True
        },
        NodeType.LOAD_BALANCER: {
            "primary": RedundancyType.ACTIVE_ACTIVE,
            "alternative": RedundancyType.ANycast,
            "min_nodes": 2,
            "target_failover_ms": 0,
            "state_sync": False
        },
        NodeType.SERVER: {
            "primary": RedundancyType.ACTIVE_ACTIVE,
            "alternative": RedundancyType.ECMP,
            "min_nodes": 3,
            "target_failover_ms": 0,
            "state_sync": False
        }
    }
    
    @staticmethod
    def plan_redundancy(
        node_type: NodeType,
        nodes: List[NetworkNode]
    ) -> RedundancyGroup:
        """
        Create a redundancy group for the given nodes.
        """
        strategy = RedundancyPlanner.RECOMMENDED_STRATEGIES.get(node_type)
        if not strategy:
            raise ValueError(f"No redundancy strategy for node type: {node_type}")
        
        if len(nodes) < strategy["min_nodes"]:
            raise ValueError(
                f"Need at least {strategy['min_nodes']} nodes for "
                f"{strategy['primary'].value} redundancy, got {len(nodes)}"
            )
        
        return RedundancyGroup(
            group_id=f"{node_type.value}-ha-group",
            redundancy_type=strategy["primary"],
            primary_node=nodes[0].node_id,
            secondary_nodes=[n.node_id for n in nodes[1:]],
            failover_time_ms=strategy["target_failover_ms"],
            state_sync_enabled=strategy["state_sync"]
        )


# Example
from io.thecodeforge.network.node_classifier import NodeRole

routers = [
    NetworkNode("rtr-01", "router-primary", NodeType.ROUTER, NodeRole.BACKBONE),
    NetworkNode("rtr-02", "router-secondary", NodeType.ROUTER, NodeRole.BACKBONE)
]

ha_group = RedundancyPlanner.plan_redundancy(NodeType.ROUTER, routers)
print(f"Redundancy type: {ha_group.redundancy_type.value}")
print(f"Nodes: {ha_group.total_nodes}")
print(f"Target failover: {ha_group.failover_time_ms}ms")

from dataclasses import dataclass, field
from typing import List, Dict, Optional
from datetime import datetime
from io.thecodeforge.network.node_classifier import NetworkNode, NodeType


@dataclass
class NodeMetrics:
    """
    Comprehensive metrics for a network node.
    """
    
    node_id: str
    timestamp: datetime
    cpu_percent: float = 0.0,
            "
    memory_percent: float = , float] = field(default_factory=dict)
    packet_loss_percent: float = 0.0
    latency_ms: float = 0.0
    error_count: int = 0
    uptime_seconds: float = 0.0
    
    @property
    def is_healthy(self) -> bool:
        return (
            self.cpu_percent < 80.0
            and self.memory_percent < 85.0
            and self.packet_loss_percent < 0.1
            and self.latency_ms < 50.0
        )
    
    @property
    def health_issues(self) -> List[str]:
        issues = []
        if self.cpu_percent >= 80.0:
            issues.append(f"CPU at {self.cpu_percent}%")
        if self.memory_percent >= 85.0:
            issues.append(f"Memory at {self.memory_percent}%")
        if self.packet_loss_percentmemory_percent": 90.0,
            "packet_loss_percent": 0.1,
            "latency_ms": 50.0.0
    interface_utilization: Dict[str >= 0.1:
            issues.append(f"Packet loss at {self.packet_loss_percent}%")
        if self.latency_ms >= 50.0:
            issues.append(f"Latency at {self.latency_ms}ms")
        return issues


class NodeMonitor:
    """
    Monitors network nodes with type-specific health checks.
    """
    
    THRESHOLDS = {
        NodeType.ROUTER: {
            "cpu_percent": 70.0,
            "memory_percent": 80.0,
            "packet_loss_percent": 0.01,
            "latency_ms": 10.0
        },
        NodeType.SWITCH: {
            "cpu_percent": 60.0,
            "memory_percent": 75.0,
            "packet_loss_percent": 0.001,
            "latency_ms": 5.0
        },
        NodeType.SERVER: {
            "cpu_percent": 85.00
        }
    }
    
    def __init__(self):
        self.metrics_history: Dict[str, List[NodeMetrics]] = {}
    
    def record_metrics(self, metrics: NodeMetrics) -> None:
        if metrics.node_id not in self.metrics_history:
            self.metrics_history[metrics.node_id] = []
        self.metrics_history[metrics.node_id].append(metrics)
    
    def check_thresholds(
        self,
        node_id: str,
        node_type: NodeType,
        metrics: NodeMetrics
    ) -> List[str]:
        """
        Check metrics against type-specific thresholds.
        """
        alerts = []
        thresholds = self.THRESHOLDS.get(node_type, {})
        
        for metric, limit in thresholds.items():
            value = getattr(metrics, metric, None)
            if value is not None and value >= limit:
                alerts.append(
                    f"{node_id}: {metric} = {value} exceeds threshold {limit}"
                )
        
        return alerts
    
    def detect_anomalies(
        self,
        node_id: str,
        window_minutes: int = 5
    ) -> List[str]:
        """
        Detect anomalies in recent metrics history.
        """
        history = self.metrics_history.get(node_id, [])
        if len(history) < 2:
            return []
        
        anomalies = []
        recent = history[-1]
        previous = history[-2]
        
        cpu_delta = abs(recent.cpu_percent - previous.cpu_percent)
        if cpu_delta > 30.0:
            anomalies.append(
                f"CPU spiked {cpu_delta:.1f}% in last interval"
            )
        
        loss_delta = abs(recent.packet_loss_percent - previous.packet_loss_percent)
        if loss_delta > 1.0:
            anomalies.append(
                f"Packet loss changed by {loss_delta:.2f}% in last interval"
            )
        
        return anomalies


# Example monitoring
monitor = NodeMonitor()
metrics = NodeMetrics(
    node_id="core-rtr-01",
    timestamp=datetime.now(),
    cpu_percent=45.0,
    memory_percent=62.0,
    packet_loss_percent=0.005,
    latency_ms=2.3
)

alerts = monitor.check_thresholds("core-rtr-01", NodeType.ROUTER, metrics)
if alerts:
    for alert in alerts:
        print(f"ALERT: {alert}")
else:
    print("All metrics within thresholds")