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

Merge remote-tracking branch 'origin/dev' into feature/2533-improve-dev-tools-in-primaite-cli

Browse Source
This commit is contained in:
Czar Echavez
2024-04-30 18:27:05 +01:00
parent ab3d23785d 13cabf8e21
commit 73f64cd89c
34 changed files with 2470 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
src/primaite/simulator/core.py
View File

@@ -1,5 +1,6 @@
# flake8: noqa
"""Core of the PrimAITE Simulator."""
import warnings
from abc import abstractmethod
from typing import Callable, Dict, List, Literal, Optional, Union
from uuid import uuid4
@@ -26,6 +27,12 @@ class RequestPermissionValidator(BaseModel):
"""Use the request and context parameters to decide whether the request should be permitted."""
pass
@property
@abstractmethod
def fail_message(self) -> str:
"""Message that is reported when a request is rejected by this validator."""
return "request rejected"
class AllowAllValidator(RequestPermissionValidator):
"""Always allows the request."""
@@ -34,6 +41,16 @@ class AllowAllValidator(RequestPermissionValidator):
"""Always allow the request."""
return True
@property
def fail_message(self) -> str:
"""
Message that is reported when a request is rejected by this validator.
This method should really never be called because this validator never rejects requests.
"""
warnings.warn("Something went wrong - AllowAllValidator rejected a request.")
return super().fail_message
class RequestType(BaseModel):
"""
@@ -99,7 +116,7 @@ class RequestManager(BaseModel):
if not request_type.validator(request_options, context):
_LOGGER.debug(f"Request {request} was denied due to insufficient permissions")
return RequestResponse(status="failure", data={"reason": "request validation failed"})
return RequestResponse(status="failure", data={"reason": request_type.validator.fail_message})
return request_type.func(request_options, context)

5
src/primaite/simulator/domain/controller.py
View File

@@ -57,6 +57,11 @@ class GroupMembershipValidator(RequestPermissionValidator):
return True
return False
@property
def fail_message(self) -> str:
"""Message that is reported when a request is rejected by this validator."""
return "User does not belong to group"
class DomainController(SimComponent):
"""Main object for controlling the domain."""

5
src/primaite/simulator/file_system/file.py
View File

@@ -3,6 +3,7 @@ from __future__ import annotations
import hashlib
import json
import os.path
import warnings
from pathlib import Path
from typing import Dict, Optional
@@ -145,6 +146,10 @@ class File(FileSystemItemABC):
Return False if corruption is detected, otherwise True
"""
warnings.warn("NODE_FILE_CHECKHASH is currently not implemented.")
self.sys_log.warning("NODE_FILE_CHECKHASH is currently not implemented.")
return False
if self.deleted:
self.sys_log.error(f"Unable to check hash of deleted file {self.folder_name}/{self.name}")
return False

2
src/primaite/simulator/file_system/file_system_item_abc.py
View File

@@ -156,7 +156,7 @@ class FileSystemItemABC(SimComponent):
@abstractmethod
def check_hash(self) -> bool:
"""
Checks the has of the file to detect any changes.
Checks the hash of the file to detect any changes.
For current implementation, any change in file hash means it is compromised.

5
src/primaite/simulator/file_system/folder.py
View File

@@ -1,5 +1,6 @@
from __future__ import annotations
import warnings
from typing import Dict, Optional
from prettytable import MARKDOWN, PrettyTable
@@ -380,6 +381,10 @@ class Folder(FileSystemItemABC):
Return False if corruption is detected, otherwise True
"""
warnings.warn("NODE_FOLDER_CHECKHASH is currently not implemented.")
self.sys_log.error("NODE_FOLDER_CHECKHASH is currently not implemented.")
return False
if self.deleted:
self.sys_log.error(f"Unable to check hash of deleted folder {self.name}")
return False

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

@@ -798,6 +798,11 @@ class Node(SimComponent):
"""Return whether the node is on or off."""
return self.node.operating_state == NodeOperatingState.ON
@property
def fail_message(self) -> str:
"""Message that is reported when a request is rejected by this validator."""
return f"Cannot perform request on node '{self.node.hostname}' because it is not turned on."
def _init_request_manager(self) -> RequestManager:
"""
Initialise the request manager.

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

@@ -330,7 +330,7 @@ class Firewall(Router):
# check if External Inbound ACL Rules permit frame
permitted, rule = self.external_inbound_acl.is_permitted(frame)
if not permitted:
self.sys_log.info(f"Frame blocked at interface {from_network_interface} by rule {rule}")
self.sys_log.info(f"Frame blocked at external inbound by rule {rule}")
return
self.software_manager.arp.add_arp_cache_entry(
ip_address=frame.ip.src_ip_address,
@@ -360,7 +360,7 @@ class Firewall(Router):
# check if External Outbound ACL Rules permit frame
permitted, rule = self.external_outbound_acl.is_permitted(frame=frame)
if not permitted:
self.sys_log.info(f"Frame blocked at interface {from_network_interface} by rule {rule}")
self.sys_log.info(f"Frame blocked at external outbound by rule {rule}")
return
self.process_frame(frame=frame, from_network_interface=from_network_interface)
@@ -380,7 +380,7 @@ class Firewall(Router):
# check if Internal Inbound ACL Rules permit frame
permitted, rule = self.internal_inbound_acl.is_permitted(frame=frame)
if not permitted:
self.sys_log.info(f"Frame blocked at interface {from_network_interface} by rule {rule}")
self.sys_log.info(f"Frame blocked at internal inbound by rule {rule}")
return
self.process_frame(frame=frame, from_network_interface=from_network_interface)
@@ -398,7 +398,7 @@ class Firewall(Router):
"""
permitted, rule = self.internal_outbound_acl.is_permitted(frame)
if not permitted:
self.sys_log.info(f"Frame blocked at interface {from_network_interface} by rule {rule}")
self.sys_log.info(f"Frame blocked at internal outbound by rule {rule}")
return
self.software_manager.arp.add_arp_cache_entry(
ip_address=frame.ip.src_ip_address,
@@ -432,7 +432,7 @@ class Firewall(Router):
# check if DMZ Inbound ACL Rules permit frame
permitted, rule = self.dmz_inbound_acl.is_permitted(frame=frame)
if not permitted:
self.sys_log.info(f"Frame blocked at interface {from_network_interface} by rule {rule}")
self.sys_log.info(f"Frame blocked at DMZ inbound by rule {rule}")
return
self.process_frame(frame=frame, from_network_interface=from_network_interface)
@@ -452,7 +452,7 @@ class Firewall(Router):
"""
permitted, rule = self.dmz_outbound_acl.is_permitted(frame)
if not permitted:
self.sys_log.info(f"Frame blocked at interface {from_network_interface} by rule {rule}")
self.sys_log.info(f"Frame blocked at DMZ outbound by rule {rule}")
return
self.software_manager.arp.add_arp_cache_entry(
ip_address=frame.ip.src_ip_address,
@@ -688,4 +688,9 @@ class Firewall(Router):
next_hop_ip_address=IPv4Address(route.get("next_hop_ip_address")),
metric=float(route.get("metric", 0)),
)
if "default_route" in cfg:
next_hop_ip_address = cfg["default_route"].get("next_hop_ip_address", None)
if next_hop_ip_address:
firewall.route_table.set_default_route_next_hop_ip_address(next_hop_ip_address)
return firewall

37
src/primaite/simulator/network/networks.py
View File

@@ -1,5 +1,9 @@
from ipaddress import IPv4Address
import yaml
from primaite import getLogger, PRIMAITE_PATHS
from primaite.game.game import PrimaiteGame
from primaite.simulator.network.container import Network
from primaite.simulator.network.hardware.nodes.host.computer import Computer
from primaite.simulator.network.hardware.nodes.host.host_node import NIC
@@ -15,6 +19,8 @@ from primaite.simulator.system.services.dns.dns_server import DNSServer
from primaite.simulator.system.services.ftp.ftp_server import FTPServer
from primaite.simulator.system.services.web_server.web_server import WebServer
_LOGGER = getLogger(__name__)
def client_server_routed() -> Network:
"""
@@ -279,3 +285,34 @@ def arcd_uc2_network() -> Network:
router_1.acl.add_rule(action=ACLAction.PERMIT, src_port=Port.HTTP, dst_port=Port.HTTP, position=3)
return network
def _get_example_network(path: str) -> Network:
try:
with open(path, "r") as file:
cfg = yaml.safe_load(file)
except FileNotFoundError:
msg = f"Failed to locate example network config {path}. Run `primaite setup` to load the example config files."
_LOGGER.error(msg)
raise FileNotFoundError(msg)
game = PrimaiteGame.from_config(cfg)
return game.simulation.network
def client_server_p2p_network_example() -> Network:
"""Get the Client-Server P2P example network."""
path = PRIMAITE_PATHS.user_config_path / "example_config" / "client_server_p2p_network_example.yaml"
return _get_example_network(path)
def basic_lan_network_example() -> Network:
"""Get the basic LAN example network."""
path = PRIMAITE_PATHS.user_config_path / "example_config" / "basic_network_network_example.yaml"
return _get_example_network(path)
def multi_lan_internet_network_example() -> Network:
"""Get Multi-LAN with Internet example network."""
path = PRIMAITE_PATHS.user_config_path / "example_config" / "multi_lan_internet_network_example.yaml"
return _get_example_network(path)