oxbrbtx/PrimAITE
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

#2248 - Initial work has been done on moving ICMP into services. still tidying up to be done. Need to fix tests too.

Browse Source
This commit is contained in:
Chris McCarthy
2024-02-02 15:35:02 +00:00
parent 1964ab4635
commit 87d9d6da04
14 changed files with 495 additions and 373 deletions
Download Patch File Download Diff File Expand all files Collapse all files

174
src/primaite/simulator/network/hardware/base.py
View File

@@ -4,7 +4,7 @@ import re
import secrets
from ipaddress import IPv4Address, IPv4Network
from pathlib import Path
from typing import Any, Dict, List, Literal, Optional, Tuple, Union
from typing import Any, Dict, List, Literal, Optional, Union
from prettytable import MARKDOWN, PrettyTable
@@ -17,7 +17,7 @@ from primaite.simulator.file_system.file_system import FileSystem
from primaite.simulator.network.hardware.node_operating_state import NodeOperatingState
from primaite.simulator.network.protocols.arp import ARPEntry, ARPPacket
from primaite.simulator.network.transmission.data_link_layer import EthernetHeader, Frame
from primaite.simulator.network.transmission.network_layer import ICMPPacket, ICMPType, IPPacket, IPProtocol
from primaite.simulator.network.transmission.network_layer import IPPacket
from primaite.simulator.network.transmission.transport_layer import Port, TCPHeader
from primaite.simulator.system.applications.application import Application
from primaite.simulator.system.core.packet_capture import PacketCapture
@@ -854,113 +854,6 @@ class ARPCache:
return item in self.arp
class ICMP:
"""
The ICMP (Internet Control Message Protocol) class.
Provides functionalities for managing and handling ICMP packets, including echo requests and replies.
"""
def __init__(self, sys_log: SysLog):
"""
Initialize the ICMP (Internet Control Message Protocol) service.
:param sys_log: The system log to store system messages and information.
:param arp_cache: The ARP cache for resolving IP to MAC address mappings.
"""
self.sys_log: SysLog = sys_log
self.software_manager: SoftwareManager = None ## noqa
self.request_replies = {}
def clear(self):
"""Clears the ICMP request replies tracker."""
self.request_replies.clear()
def process_icmp(self, frame: Frame, from_nic: NIC, is_reattempt: bool = False):
"""
Process an ICMP packet, including handling echo requests and replies.
:param frame: The Frame containing the ICMP packet to process.
"""
if frame.icmp.icmp_type == ICMPType.ECHO_REQUEST:
if not is_reattempt:
self.sys_log.info(f"Received echo request from {frame.ip.src_ip_address}")
target_mac_address = self.software_manager.arp.get_arp_cache_mac_address(frame.ip.src_ip_address)
src_nic = self.software_manager.arp.get_arp_cache_nic(frame.ip.src_ip_address)
if not src_nic:
self.software_manager.arp.send_arp_request(frame.ip.src_ip_address)
self.process_icmp(frame=frame, from_nic=from_nic, is_reattempt=True)
return
# Network Layer
ip_packet = IPPacket(
src_ip_address=src_nic.ip_address, dst_ip_address=frame.ip.src_ip_address, protocol=IPProtocol.ICMP
)
# Data Link Layer
ethernet_header = EthernetHeader(src_mac_addr=src_nic.mac_address, dst_mac_addr=target_mac_address)
icmp_reply_packet = ICMPPacket(
icmp_type=ICMPType.ECHO_REPLY,
icmp_code=0,
identifier=frame.icmp.identifier,
sequence=frame.icmp.sequence + 1,
)
payload = secrets.token_urlsafe(int(32 / 1.3)) # Standard ICMP 32 bytes size
frame = Frame(ethernet=ethernet_header, ip=ip_packet, icmp=icmp_reply_packet, payload=payload)
self.sys_log.info(f"Sending echo reply to {frame.ip.dst_ip_address}")
src_nic.send_frame(frame)
elif frame.icmp.icmp_type == ICMPType.ECHO_REPLY:
time = frame.transmission_duration()
time_str = f"{time}ms" if time > 0 else "<1ms"
self.sys_log.info(
f"Reply from {frame.ip.src_ip_address}: "
f"bytes={len(frame.payload)}, "
f"time={time_str}, "
f"TTL={frame.ip.ttl}"
)
if not self.request_replies.get(frame.icmp.identifier):
self.request_replies[frame.icmp.identifier] = 0
self.request_replies[frame.icmp.identifier] += 1
def ping(
self, target_ip_address: IPv4Address, sequence: int = 0, identifier: Optional[int] = None, pings: int = 4
) -> Tuple[int, Union[int, None]]:
"""
Send an ICMP echo request (ping) to a target IP address and manage the sequence and identifier.
:param target_ip_address: The target IP address to send the ping.
:param sequence: The sequence number of the echo request. Defaults to 0.
:param identifier: An optional identifier for the ICMP packet. If None, a default will be used.
:return: A tuple containing the next sequence number and the identifier, or (0, None) if the target IP address
was not found in the ARP cache.
"""
nic = self.software_manager.arp.get_arp_cache_nic(target_ip_address)
if not nic:
return pings, None
# ARP entry exists
sequence += 1
target_mac_address = self.software_manager.arp.get_arp_cache_mac_address(target_ip_address)
src_nic = self.software_manager.arp.get_arp_cache_nic(target_ip_address)
tcp_header = TCPHeader(src_port=Port.ARP, dst_port=Port.ARP)
# Network Layer
ip_packet = IPPacket(
src_ip_address=nic.ip_address,
dst_ip_address=target_ip_address,
protocol=IPProtocol.ICMP,
)
# Data Link Layer
ethernet_header = EthernetHeader(src_mac_addr=src_nic.mac_address, dst_mac_addr=target_mac_address)
icmp_packet = ICMPPacket(identifier=identifier, sequence=sequence)
payload = secrets.token_urlsafe(int(32 / 1.3)) # Standard ICMP 32 bytes size
frame = Frame(ethernet=ethernet_header, ip=ip_packet, tcp=tcp_header, icmp=icmp_packet, payload=payload)
nic.send_frame(frame)
return sequence, icmp_packet.identifier
class Node(SimComponent):
"""
@@ -999,7 +892,6 @@ class Node(SimComponent):
root: Path
"Root directory for simulation output."
sys_log: SysLog
icmp: ICMP
session_manager: SessionManager
software_manager: SoftwareManager
@@ -1042,8 +934,6 @@ class Node(SimComponent):
kwargs["default_gateway"] = IPv4Address(kwargs["default_gateway"])
if not kwargs.get("sys_log"):
kwargs["sys_log"] = SysLog(kwargs["hostname"])
if not kwargs.get("icmp"):
kwargs["icmp"] = ICMP(sys_log=kwargs.get("sys_log"))
if not kwargs.get("session_manager"):
kwargs["session_manager"] = SessionManager(sys_log=kwargs.get("sys_log"), arp_cache=kwargs.get("arp"))
if not kwargs.get("root"):
@@ -1059,7 +949,6 @@ class Node(SimComponent):
dns_server=kwargs.get("dns_server"),
)
super().__init__(**kwargs)
self.icmp.software_manager = self.software_manager
self.session_manager.node = self
self.session_manager.software_manager = self.software_manager
self._install_system_software()
@@ -1096,12 +985,6 @@ class Node(SimComponent):
"""Reset the original state of the SimComponent."""
super().reset_component_for_episode(episode)
# Reset ARP Cache
self.arp.clear()
# Reset ICMP
self.icmp.clear()
# Reset Session Manager
self.session_manager.clear()
@@ -1436,35 +1319,9 @@ class Node(SimComponent):
:param pings: The number of pings to attempt, default is 4.
:return: True if the ping is successful, otherwise False.
"""
if self.operating_state == NodeOperatingState.ON:
if not isinstance(target_ip_address, IPv4Address):
target_ip_address = IPv4Address(target_ip_address)
if target_ip_address.is_loopback:
self.sys_log.info("Pinging loopback address")
return any(nic.enabled for nic in self.nics.values())
if self.operating_state == NodeOperatingState.ON:
output = f"Pinging {target_ip_address}:"
self.sys_log.info(output)
print(output)
sequence, identifier = 0, None
while sequence < pings:
sequence, identifier = self.icmp.ping(target_ip_address, sequence, identifier, pings)
request_replies = self.icmp.request_replies.get(identifier)
passed = request_replies == pings
if request_replies:
self.icmp.request_replies.pop(identifier)
else:
request_replies = 0
output = (
f"Ping statistics for {target_ip_address}: "
f"Packets: Sent = {pings}, "
f"Received = {request_replies}, "
f"Lost = {pings - request_replies} ({(pings - request_replies) / pings * 100}% loss)"
)
self.sys_log.info(output)
print(output)
return passed
return False
if not isinstance(target_ip_address, IPv4Address):
target_ip_address = IPv4Address(target_ip_address)
return self.software_manager.icmp.ping(target_ip_address)
def send_frame(self, frame: Frame):
"""
@@ -1492,22 +1349,23 @@ class Node(SimComponent):
self.software_manager.arp.add_arp_cache_entry(
ip_address=frame.ip.src_ip_address, mac_address=frame.ethernet.src_mac_addr, nic=from_nic
)
if frame.ip.protocol == IPProtocol.ICMP:
self.icmp.process_icmp(frame=frame, from_nic=from_nic)
return
# Check if the destination port is open on the Node
dst_port = None
if frame.tcp:
dst_port = frame.tcp.dst_port
elif frame.udp:
dst_port = frame.udp.dst_port
if dst_port in self.software_manager.get_open_ports():
# accept thr frame as the port is open
accept_frame = False
if frame.icmp or dst_port in self.software_manager.get_open_ports():
# accept the frame as the port is open or if it's an ICMP frame
accept_frame = True
# TODO: add internal node firewall check here?
if accept_frame:
self.session_manager.receive_frame(frame, from_nic)
# if frame.tcp.src_port == Port.ARP:
# self.arp.process_arp_packet(from_nic=from_nic, arp_packet=frame.arp)
# else:
# self.session_manager.receive_frame(frame)
else:
# denied as port closed
self.sys_log.info(f"Ignoring frame for port {frame.tcp.dst_port.value} from {frame.ip.src_ip_address}")
@@ -1527,7 +1385,6 @@ class Node(SimComponent):
self.services[service.uuid] = service
service.parent = self
service.install() # Perform any additional setup, such as creating files for this service on the node.
self.sys_log.info(f"Installed service {service.name}")
self._service_request_manager.add_request(service.uuid, RequestType(func=service._request_manager))
def uninstall_service(self, service: Service) -> None:
@@ -1559,7 +1416,6 @@ class Node(SimComponent):
return
self.applications[application.uuid] = application
application.parent = self
self.sys_log.info(f"Installed application {application.name}")
self._application_request_manager.add_request(application.uuid, RequestType(func=application._request_manager))
def uninstall_application(self, application: Application) -> None:

67
src/primaite/simulator/network/hardware/nodes/host.py Normal file
View File

@@ -0,0 +1,67 @@
from primaite.simulator.network.hardware.base import NIC, Node
from primaite.simulator.system.applications.web_browser import WebBrowser
from primaite.simulator.system.services.arp.host_arp import HostARP
from primaite.simulator.system.services.dns.dns_client import DNSClient
from primaite.simulator.system.services.ftp.ftp_client import FTPClient
from primaite.simulator.system.services.icmp.icmp import ICMP
from primaite.simulator.system.services.ntp.ntp_client import NTPClient
class Host(Node):
"""
A basic Host class.
Example:
>>> pc_a = Host(
hostname="pc_a",
ip_address="192.168.1.10",
subnet_mask="255.255.255.0",
default_gateway="192.168.1.1"
)
>>> pc_a.power_on()
Instances of computer come 'pre-packaged' with the following:
* Core Functionality:
* ARP
* ICMP
* Packet Capture
* Sys Log
* Services:
* DNS Client
* FTP Client
* LDAP Client
* NTP Client
* Applications:
* Email Client
* Web Browser
* Processes:
* Placeholder
"""
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.connect_nic(NIC(ip_address=kwargs["ip_address"], subnet_mask=kwargs["subnet_mask"]))
self._install_system_software()
def _install_system_software(self):
"""Install System Software - software that is usually provided with the OS."""
# ARP Service
self.software_manager.install(HostARP)
# ICMP Service
self.software_manager.install(ICMP)
# DNS Client
self.software_manager.install(DNSClient)
# FTP Client
self.software_manager.install(FTPClient)
# NTP Client
self.software_manager.install(NTPClient)
# Web Browser
self.software_manager.install(WebBrowser)
super()._install_system_software()

99
src/primaite/simulator/network/hardware/nodes/router.py
View File

@@ -8,10 +8,10 @@ from typing import Dict, List, Optional, Tuple, Union
from prettytable import MARKDOWN, PrettyTable
from primaite.simulator.core import RequestManager, RequestType, SimComponent
from primaite.simulator.network.hardware.base import ARPCache, ICMP, NIC, Node
from primaite.simulator.network.hardware.base import ARPCache, NIC, Node
from primaite.simulator.network.hardware.node_operating_state import NodeOperatingState
from primaite.simulator.network.transmission.data_link_layer import EthernetHeader, Frame
from primaite.simulator.network.transmission.network_layer import ICMPPacket, ICMPType, IPPacket, IPProtocol
from primaite.simulator.network.transmission.network_layer import IPPacket, IPProtocol
from primaite.simulator.network.transmission.transport_layer import Port, TCPHeader
from primaite.simulator.system.core.sys_log import SysLog
@@ -628,96 +628,6 @@ class RouterARPCache(ARPCache):
# return
class RouterICMP(ICMP):
"""
A class to represent a router's Internet Control Message Protocol (ICMP) handler.
:param sys_log: System log for logging network events and errors.
:type sys_log: SysLog
:param arp_cache: The ARP cache for resolving MAC addresses.
:type arp_cache: ARPCache
:param router: The router to which this ICMP handler belongs.
:type router: Router
"""
router: Router
def __init__(self, sys_log: SysLog, arp_cache: ARPCache, router: Router):
super().__init__(sys_log, arp_cache)
self.router = router
def process_icmp(self, frame: Frame, from_nic: NIC, is_reattempt: bool = False):
"""
Process incoming ICMP frames based on ICMP type.
:param frame: The incoming frame to process.
:param from_nic: The network interface where the frame is coming from.
:param is_reattempt: Flag to indicate if the process is a reattempt.
"""
if frame.icmp.icmp_type == ICMPType.ECHO_REQUEST:
# determine if request is for router interface or whether it needs to be routed
for nic in self.router.nics.values():
if nic.ip_address == frame.ip.dst_ip_address:
if nic.enabled:
# reply to the request
if not is_reattempt:
self.sys_log.info(f"Received echo request from {frame.ip.src_ip_address}")
target_mac_address = self.arp.get_arp_cache_mac_address(frame.ip.src_ip_address)
src_nic = self.arp.get_arp_cache_nic(frame.ip.src_ip_address)
tcp_header = TCPHeader(src_port=Port.ARP, dst_port=Port.ARP)
# Network Layer
ip_packet = IPPacket(
src_ip_address=nic.ip_address,
dst_ip_address=frame.ip.src_ip_address,
protocol=IPProtocol.ICMP,
)
# Data Link Layer
ethernet_header = EthernetHeader(
src_mac_addr=src_nic.mac_address, dst_mac_addr=target_mac_address
)
icmp_reply_packet = ICMPPacket(
icmp_type=ICMPType.ECHO_REPLY,
icmp_code=0,
identifier=frame.icmp.identifier,
sequence=frame.icmp.sequence + 1,
)
payload = secrets.token_urlsafe(int(32 / 1.3)) # Standard ICMP 32 bytes size
frame = Frame(
ethernet=ethernet_header,
ip=ip_packet,
tcp=tcp_header,
icmp=icmp_reply_packet,
payload=payload,
)
self.sys_log.info(f"Sending echo reply to {frame.ip.dst_ip_address}")
src_nic.send_frame(frame)
return
# Route the frame
self.router.process_frame(frame, from_nic)
elif frame.icmp.icmp_type == ICMPType.ECHO_REPLY:
for nic in self.router.nics.values():
if nic.ip_address == frame.ip.dst_ip_address:
if nic.enabled:
time = frame.transmission_duration()
time_str = f"{time}ms" if time > 0 else "<1ms"
self.sys_log.info(
f"Reply from {frame.ip.src_ip_address}: "
f"bytes={len(frame.payload)}, "
f"time={time_str}, "
f"TTL={frame.ip.ttl}"
)
if not self.request_replies.get(frame.icmp.identifier):
self.request_replies[frame.icmp.identifier] = 0
self.request_replies[frame.icmp.identifier] += 1
return
# Route the frame
self.router.process_frame(frame, from_nic)
class RouterNIC(NIC):
@@ -786,9 +696,10 @@ class Router(Node):
kwargs["route_table"] = RouteTable(sys_log=kwargs["sys_log"])
if not kwargs.get("arp"):
kwargs["arp"] = RouterARPCache(sys_log=kwargs.get("sys_log"), router=self)
if not kwargs.get("icmp"):
kwargs["icmp"] = RouterICMP(sys_log=kwargs.get("sys_log"), arp_cache=kwargs.get("arp"), router=self)
# if not kwargs.get("icmp"):
# kwargs["icmp"] = RouterICMP(sys_log=kwargs.get("sys_log"), arp_cache=kwargs.get("arp"), router=self)
super().__init__(hostname=hostname, num_ports=num_ports, **kwargs)
# TODO: Install RoputerICMP
for i in range(1, self.num_ports + 1):
nic = RouterNIC(ip_address="127.0.0.1", subnet_mask="255.0.0.0", gateway="0.0.0.0")
self.connect_nic(nic)

114
src/primaite/simulator/network/protocols/icmp.py Normal file
View File

@@ -0,0 +1,114 @@
import secrets
from enum import Enum
from typing import Union
from pydantic import BaseModel, field_validator, validate_call
from pydantic_core.core_schema import FieldValidationInfo
from primaite import getLogger
_LOGGER = getLogger(__name__)
class ICMPType(Enum):
"""Enumeration of common ICMP (Internet Control Message Protocol) types."""
ECHO_REPLY = 0
"Echo Reply message."
DESTINATION_UNREACHABLE = 3
"Destination Unreachable."
REDIRECT = 5
"Redirect."
ECHO_REQUEST = 8
"Echo Request (ping)."
ROUTER_ADVERTISEMENT = 10
"Router Advertisement."
ROUTER_SOLICITATION = 11
"Router discovery/selection/solicitation."
TIME_EXCEEDED = 11
"Time Exceeded."
TIMESTAMP_REQUEST = 13
"Timestamp Request."
TIMESTAMP_REPLY = 14
"Timestamp Reply."
@validate_call
def get_icmp_type_code_description(icmp_type: ICMPType, icmp_code: int) -> Union[str, None]:
"""
Maps ICMPType and code pairings to their respective description.
:param icmp_type: An ICMPType.
:param icmp_code: An icmp code.
:return: The icmp type and code pairing description if it exists, otherwise returns None.
"""
icmp_code_descriptions = {
ICMPType.ECHO_REPLY: {0: "Echo reply"},
ICMPType.DESTINATION_UNREACHABLE: {
0: "Destination network unreachable",
1: "Destination host unreachable",
2: "Destination protocol unreachable",
3: "Destination port unreachable",
4: "Fragmentation required",
5: "Source route failed",
6: "Destination network unknown",
7: "Destination host unknown",
8: "Source host isolated",
9: "Network administratively prohibited",
10: "Host administratively prohibited",
11: "Network unreachable for ToS",
12: "Host unreachable for ToS",
13: "Communication administratively prohibited",
14: "Host Precedence Violation",
15: "Precedence cutoff in effect",
},
ICMPType.REDIRECT: {
0: "Redirect Datagram for the Network",
1: "Redirect Datagram for the Host",
},
ICMPType.ECHO_REQUEST: {0: "Echo request"},
ICMPType.ROUTER_ADVERTISEMENT: {0: "Router Advertisement"},
ICMPType.ROUTER_SOLICITATION: {0: "Router discovery/selection/solicitation"},
ICMPType.TIME_EXCEEDED: {0: "TTL expired in transit", 1: "Fragment reassembly time exceeded"},
ICMPType.TIMESTAMP_REQUEST: {0: "Timestamp Request"},
ICMPType.TIMESTAMP_REPLY: {0: "Timestamp reply"},
}
return icmp_code_descriptions[icmp_type].get(icmp_code)
class ICMPPacket(BaseModel):
"""Models an ICMP Packet."""
icmp_type: ICMPType = ICMPType.ECHO_REQUEST
"ICMP Type."
icmp_code: int = 0
"ICMP Code."
identifier: int
"ICMP identifier (16 bits randomly generated)."
sequence: int = 0
"ICMP message sequence number."
def __init__(self, **kwargs):
if not kwargs.get("identifier"):
kwargs["identifier"] = secrets.randbits(16)
super().__init__(**kwargs)
@field_validator("icmp_code") # noqa
@classmethod
def _icmp_type_must_have_icmp_code(cls, v: int, info: FieldValidationInfo) -> int:
"""Validates the icmp_type and icmp_code."""
icmp_type = info.data["icmp_type"]
if get_icmp_type_code_description(icmp_type, v):
return v
msg = f"No Matching ICMP code for type:{icmp_type.name}, code:{v}"
_LOGGER.error(msg)
raise ValueError(msg)
def code_description(self) -> str:
"""The icmp_code description."""
description = get_icmp_type_code_description(self.icmp_type, self.icmp_code)
if description:
return description
msg = f"No Matching ICMP code for type:{self.icmp_type.name}, code:{self.icmp_code}"
_LOGGER.error(msg)
raise ValueError(msg)

3
src/primaite/simulator/network/transmission/data_link_layer.py
View File

@@ -5,8 +5,9 @@ from pydantic import BaseModel
from primaite import getLogger
from primaite.simulator.network.protocols.arp import ARPPacket
from primaite.simulator.network.protocols.icmp import ICMPPacket
from primaite.simulator.network.protocols.packet import DataPacket
from primaite.simulator.network.transmission.network_layer import ICMPPacket, IPPacket, IPProtocol
from primaite.simulator.network.transmission.network_layer import IPPacket, IPProtocol
from primaite.simulator.network.transmission.primaite_layer import PrimaiteHeader
from primaite.simulator.network.transmission.transport_layer import TCPHeader, UDPHeader
from primaite.simulator.network.utils import convert_bytes_to_megabits

104
src/primaite/simulator/network/transmission/network_layer.py
View File

@@ -54,110 +54,6 @@ class Precedence(Enum):
"Highest priority level, used for the most critical network control messages, such as routing protocol hellos."
class ICMPType(Enum):
"""Enumeration of common ICMP (Internet Control Message Protocol) types."""
ECHO_REPLY = 0
"Echo Reply message."
DESTINATION_UNREACHABLE = 3
"Destination Unreachable."
REDIRECT = 5
"Redirect."
ECHO_REQUEST = 8
"Echo Request (ping)."
ROUTER_ADVERTISEMENT = 10
"Router Advertisement."
ROUTER_SOLICITATION = 11
"Router discovery/selection/solicitation."
TIME_EXCEEDED = 11
"Time Exceeded."
TIMESTAMP_REQUEST = 13
"Timestamp Request."
TIMESTAMP_REPLY = 14
"Timestamp Reply."
@validate_call
def get_icmp_type_code_description(icmp_type: ICMPType, icmp_code: int) -> Union[str, None]:
"""
Maps ICMPType and code pairings to their respective description.
:param icmp_type: An ICMPType.
:param icmp_code: An icmp code.
:return: The icmp type and code pairing description if it exists, otherwise returns None.
"""
icmp_code_descriptions = {
ICMPType.ECHO_REPLY: {0: "Echo reply"},
ICMPType.DESTINATION_UNREACHABLE: {
0: "Destination network unreachable",
1: "Destination host unreachable",
2: "Destination protocol unreachable",
3: "Destination port unreachable",
4: "Fragmentation required",
5: "Source route failed",
6: "Destination network unknown",
7: "Destination host unknown",
8: "Source host isolated",
9: "Network administratively prohibited",
10: "Host administratively prohibited",
11: "Network unreachable for ToS",
12: "Host unreachable for ToS",
13: "Communication administratively prohibited",
14: "Host Precedence Violation",
15: "Precedence cutoff in effect",
},
ICMPType.REDIRECT: {
0: "Redirect Datagram for the Network",
1: "Redirect Datagram for the Host",
},
ICMPType.ECHO_REQUEST: {0: "Echo request"},
ICMPType.ROUTER_ADVERTISEMENT: {0: "Router Advertisement"},
ICMPType.ROUTER_SOLICITATION: {0: "Router discovery/selection/solicitation"},
ICMPType.TIME_EXCEEDED: {0: "TTL expired in transit", 1: "Fragment reassembly time exceeded"},
ICMPType.TIMESTAMP_REQUEST: {0: "Timestamp Request"},
ICMPType.TIMESTAMP_REPLY: {0: "Timestamp reply"},
}
return icmp_code_descriptions[icmp_type].get(icmp_code)
class ICMPPacket(BaseModel):
"""Models an ICMP Packet."""
icmp_type: ICMPType = ICMPType.ECHO_REQUEST
"ICMP Type."
icmp_code: int = 0
"ICMP Code."
identifier: int
"ICMP identifier (16 bits randomly generated)."
sequence: int = 0
"ICMP message sequence number."
def __init__(self, **kwargs):
if not kwargs.get("identifier"):
kwargs["identifier"] = secrets.randbits(16)
super().__init__(**kwargs)
@field_validator("icmp_code") # noqa
@classmethod
def _icmp_type_must_have_icmp_code(cls, v: int, info: FieldValidationInfo) -> int:
"""Validates the icmp_type and icmp_code."""
icmp_type = info.data["icmp_type"]
if get_icmp_type_code_description(icmp_type, v):
return v
msg = f"No Matching ICMP code for type:{icmp_type.name}, code:{v}"
_LOGGER.error(msg)
raise ValueError(msg)
def code_description(self) -> str:
"""The icmp_code description."""
description = get_icmp_type_code_description(self.icmp_type, self.icmp_code)
if description:
return description
msg = f"No Matching ICMP code for type:{self.icmp_type.name}, code:{self.icmp_code}"
_LOGGER.error(msg)
raise ValueError(msg)
class IPPacket(BaseModel):
"""
Represents the IP layer of a network frame.

2
src/primaite/simulator/network/transmission/transport_layer.py
View File

@@ -7,6 +7,8 @@ from pydantic import BaseModel
class Port(Enum):
"""Enumeration of common known TCP/UDP ports used by protocols for operation of network applications."""
NONE = 0
"Place holder for a non-port."
WOL = 9
"Wake-on-Lan (WOL) - Used to turn or awaken a computer from sleep mode by a network message."
FTP_DATA = 20