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

#3110: update port and protocol descriptions

Browse Source
This commit is contained in:
Nick Todd
2025-03-11 16:41:32 +00:00
parent dab261da0b
commit d2b5097dce
13 changed files with 74 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/primaite/simulator/network/hardware/nodes/network/firewall.py
View File

@@ -57,8 +57,8 @@ class Firewall(Router, discriminator="firewall"):
>>> # Permit HTTP traffic to the DMZ
>>> firewall.dmz_inbound_acl.add_rule(
... action=ACLAction.PERMIT,
... protocol=IPProtocol["TCP"],
... dst_port=Port["HTTP"],
... protocol=PROTOCOL_LOOKUP["TCP"],
... dst_port=PORT_LOOKUP["HTTP"],
... src_ip_address="0.0.0.0",
... src_wildcard_mask="0.0.0.0",
... dst_ip_address="172.16.0.0",

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

@@ -265,7 +265,7 @@ class AccessControlList(SimComponent):
>>> acl = AccessControlList()
>>> acl.add_rule(
... action=ACLAction.PERMIT,
... protocol=IPProtocol["TCP"],
... protocol=PROTOCOL_LOOKUP["TCP"],
... src_ip_address="192.168.1.0",
... src_wildcard_mask="0.0.0.255",
... dst_ip_address="192.168.2.0",
@@ -399,11 +399,11 @@ class AccessControlList(SimComponent):
>>> router = Router("router")
>>> router.add_rule(
... action=ACLAction.DENY,
... protocol=IPProtocol["TCP"],
... protocol=PROTOCOL_LOOKUP["TCP"],
... src_ip_address="192.168.1.0",
... src_wildcard_mask="0.0.0.255",
... dst_ip_address="10.10.10.5",
... dst_port=Port["SSH"],
... dst_port=PORT_LOOKUP["SSH"],
... position=5
... )
>>> # This permits SSH traffic from the 192.168.1.0/24 subnet to the 10.10.10.5 server.
@@ -411,10 +411,10 @@ class AccessControlList(SimComponent):
>>> # Then if we want to allow a specific IP address from this subnet to SSH into the server
>>> router.add_rule(
... action=ACLAction.PERMIT,
... protocol=IPProtocol["TCP"],
... protocol=PROTOCOL_LOOKUP["TCP"],
... src_ip_address="192.168.1.25",
... dst_ip_address="10.10.10.5",
... dst_port=Port["SSH"],
... dst_port=PORT_LOOKUP["SSH"],
... position=4
... )

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

@@ -61,7 +61,7 @@ class IPPacket(BaseModel):
>>> ip_packet = IPPacket(
... src_ip_address=IPv4Address('192.168.0.1'),
... dst_ip_address=IPv4Address('10.0.0.1'),
... protocol=IPProtocol["TCP"],
... protocol=PROTOCOL_LOOKUP["TCP"],
... ttl=64,
... precedence=Precedence.CRITICAL
... )

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

@@ -15,8 +15,8 @@ class UDPHeader(BaseModel):
:Example:
>>> udp_header = UDPHeader(
... src_port=Port["HTTP_ALT"],
... dst_port=Port["HTTP"],
... src_port=PORT_LOOKUP["HTTP_ALT"],
... dst_port=PORT_LOOKUP["HTTP"],
... )
"""
@@ -54,8 +54,8 @@ class TCPHeader(BaseModel):
:Example:
>>> tcp_header = TCPHeader(
... src_port=Port["HTTP_ALT"],
... dst_port=Port["HTTP"],
... src_port=PORT_LOOKUP["HTTP_ALT"],
... dst_port=PORT_LOOKUP["HTTP"],
... flags=[TCPFlags.SYN, TCPFlags.ACK]
... )
"""

2
src/primaite/simulator/system/applications/red_applications/dos_bot.py
View File

@@ -128,7 +128,7 @@ class DoSBot(DatabaseClient, discriminator="dos-bot"):
Configure the Denial of Service bot.
:param: target_ip_address: The IP address of the Node containing the target service.
:param: target_port: The port of the target service. Optional - Default is `Port["HTTP"]`
:param: target_port: The port of the target service. Optional - Default is `PORT_LOOKUP["HTTP"]`
:param: payload: The payload the DoS Bot will throw at the target service. Optional - Default is `None`
:param: repeat: If True, the bot will maintain the attack. Optional - Default is `True`
:param: port_scan_p_of_success: The chance of the port scan being successful. Optional - Default is 0.1 (10%)

2
src/primaite/simulator/system/core/session_manager.py
View File

@@ -332,7 +332,7 @@ class SessionManager:
)
# TODO: Only create IP packet if not ARP
# ip_packet = None
# if dst_port != Port["ARP"]:
# if dst_port != PORT_LOOKUP["ARP"]:
# IPPacket(
# src_ip_address=outbound_network_interface.ip_address,
# dst_ip_address=dst_ip_address,

4
src/primaite/simulator/system/services/ftp/ftp_client.py
View File

@@ -215,7 +215,7 @@ class FTPClient(FTPServiceABC, discriminator="ftp-client"):
:param: dest_file_name: The name of the file to be saved on the FTP Server.
:type: dest_file_name: str
:param: dest_port: The open port of the machine that hosts the FTP Server. Default is Port["FTP"].
:param: dest_port: The open port of the machine that hosts the FTP Server. Default is PORT_LOOKUP["FTP"].
:type: dest_port: Optional[Port]
:param: session_id: The id of the session
@@ -276,7 +276,7 @@ class FTPClient(FTPServiceABC, discriminator="ftp-client"):
:param: dest_file_name: The name of the file to be saved on the FTP Server.
:type: dest_file_name: str
:param: dest_port: The open port of the machine that hosts the FTP Server. Default is Port["FTP"].
:param: dest_port: The open port of the machine that hosts the FTP Server. Default is PORT_LOOKUP["FTP"].
:type: dest_port: Optional[int]
"""
self._active = True

2
src/primaite/simulator/system/services/ftp/ftp_service.py
View File

@@ -114,7 +114,7 @@ class FTPServiceABC(Service, ABC):
:param: dest_ip_address: The IP address of the machine that hosts the FTP Server.
:type: dest_ip_address: Optional[IPv4Address]
:param: dest_port: The open port of the machine that hosts the FTP Server. Default is Port["FTP"].
:param: dest_port: The open port of the machine that hosts the FTP Server. Default is PORT_LOOKUP["FTP"].
:type: dest_port: Optional[Port]
:param: session_id: session ID linked to the FTP Packet. Optional.

91
src/primaite/simulator/system/services/icmp/router_icmp.py
View File

@@ -1,91 +0,0 @@
# © Crown-owned copyright 2025, Defence Science and Technology Laboratory UK
# 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_network_interface: NIC, is_reattempt: bool = False):
# """
# Process incoming icmp frames based on icmp type.
#
# :param frame: The incoming frame to process.
# :param from_network_interface: 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 network_interface in self.router.network_interfaces.values():
# if network_interface.ip_address == frame.ip.dst_ip_address:
# if network_interface.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_network_interface(frame.ip.src_ip_address)
# tcp_header = TCPHeader(src_port=Port["arp"], dst_port=Port["arp"])
#
# # Network Layer
# ip_packet = IPPacket(
# src_ip_address=network_interface.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_network_interface)
#
# elif frame.icmp.icmp_type == ICMPType.ECHO_REPLY:
# for network_interface in self.router.network_interfaces.values():
# if network_interface.ip_address == frame.ip.dst_ip_address:
# if network_interface.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_network_interface)