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

#2257: apply PR suggestions

Browse Source
This commit is contained in:
Czar Echavez
2024-02-28 15:08:00 +00:00
parent f2d7a2fc16
commit 6d43c61058
11 changed files with 81 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/primaite/game/agent/rewards.py
View File

@@ -13,7 +13,7 @@ the structure:
- type: DATABASE_FILE_INTEGRITY
weight: 0.5
options:
node_ref: database_server
node_name: database_server
folder_name: database
file_name: database.db
@@ -21,7 +21,7 @@ the structure:
- type: WEB_SERVER_404_PENALTY
weight: 0.5
options:
node_ref: web_server
node_name: web_server
service_ref: web_server_database_client
```
"""
@@ -184,7 +184,7 @@ class WebServer404Penalty(AbstractReward):
service_name = config.get("service_name")
if not (node_hostname and service_name):
msg = (
f"{cls.__name__} could not be initialised from config because node_ref and service_ref were not "
f"{cls.__name__} could not be initialised from config because node_name and service_ref were not "
"found in reward config."
)
_LOGGER.warning(msg)

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

@@ -497,66 +497,66 @@ class Firewall(Router):
@classmethod
def from_config(cls, cfg: dict) -> "Firewall":
"""Create a firewall based on a config dict."""
new = Firewall(hostname=cfg["hostname"], operating_state=NodeOperatingState.ON)
firewall = Firewall(hostname=cfg["hostname"], operating_state=NodeOperatingState.ON)
if "ports" in cfg:
internal_port = cfg["ports"]["internal_port"]
external_port = cfg["ports"]["external_port"]
dmz_port = cfg["ports"]["dmz_port"]
# configure internal port
new.configure_internal_port(
firewall.configure_internal_port(
ip_address=IPV4Address(internal_port.get("ip_address")),
subnet_mask=IPV4Address(internal_port.get("subnet_mask", "255.255.255.0")),
)
# configure external port
new.configure_external_port(
firewall.configure_external_port(
ip_address=IPV4Address(external_port.get("ip_address")),
subnet_mask=IPV4Address(external_port.get("subnet_mask", "255.255.255.0")),
)
# configure dmz port
new.configure_dmz_port(
firewall.configure_dmz_port(
ip_address=IPV4Address(dmz_port.get("ip_address")),
subnet_mask=IPV4Address(dmz_port.get("subnet_mask", "255.255.255.0")),
)
if "acl" in cfg:
# acl rules for internal_inbound_acl
if cfg["acl"]["internal_inbound_acl"]:
new.internal_inbound_acl.max_acl_rules
new.internal_inbound_acl._default_config = cfg["acl"]["internal_inbound_acl"]
new.internal_inbound_acl._reset_rules_to_default()
firewall.internal_inbound_acl.max_acl_rules
firewall.internal_inbound_acl._default_config = cfg["acl"]["internal_inbound_acl"]
firewall.internal_inbound_acl._reset_rules_to_default()
# acl rules for internal_outbound_acl
if cfg["acl"]["internal_outbound_acl"]:
new.internal_outbound_acl._default_config = cfg["acl"]["internal_outbound_acl"]
new.internal_outbound_acl._reset_rules_to_default()
firewall.internal_outbound_acl._default_config = cfg["acl"]["internal_outbound_acl"]
firewall.internal_outbound_acl._reset_rules_to_default()
# acl rules for dmz_inbound_acl
if cfg["acl"]["dmz_inbound_acl"]:
new.dmz_inbound_acl._default_config = cfg["acl"]["dmz_inbound_acl"]
new.dmz_inbound_acl._reset_rules_to_default()
firewall.dmz_inbound_acl._default_config = cfg["acl"]["dmz_inbound_acl"]
firewall.dmz_inbound_acl._reset_rules_to_default()
# acl rules for dmz_outbound_acl
if cfg["acl"]["dmz_outbound_acl"]:
new.dmz_outbound_acl._default_config = cfg["acl"]["dmz_outbound_acl"]
new.dmz_outbound_acl._reset_rules_to_default()
firewall.dmz_outbound_acl._default_config = cfg["acl"]["dmz_outbound_acl"]
firewall.dmz_outbound_acl._reset_rules_to_default()
# acl rules for external_inbound_acl
if cfg["acl"]["external_inbound_acl"]:
new.external_inbound_acl._default_config = cfg["acl"]["external_inbound_acl"]
new.external_inbound_acl._reset_rules_to_default()
firewall.external_inbound_acl._default_config = cfg["acl"]["external_inbound_acl"]
firewall.external_inbound_acl._reset_rules_to_default()
# acl rules for external_outbound_acl
if cfg["acl"]["external_outbound_acl"]:
new.external_outbound_acl._default_config = cfg["acl"]["external_outbound_acl"]
new.external_outbound_acl._reset_rules_to_default()
firewall.external_outbound_acl._default_config = cfg["acl"]["external_outbound_acl"]
firewall.external_outbound_acl._reset_rules_to_default()
if "routes" in cfg:
for route in cfg.get("routes"):
new.route_table.add_route(
firewall.route_table.add_route(
address=IPv4Address(route.get("address")),
subnet_mask=IPv4Address(route.get("subnet_mask", "255.255.255.0")),
next_hop_ip_address=IPv4Address(route.get("next_hop_ip_address")),
metric=float(route.get("metric", 0)),
)
return new
return firewall

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

@@ -1480,27 +1480,27 @@ class Router(NetworkNode):
:return: Configured router.
:rtype: Router
"""
new = Router(
router = Router(
hostname=cfg["hostname"],
num_ports=int(cfg.get("num_ports", "5")),
operating_state=NodeOperatingState.ON,
)
if "ports" in cfg:
for port_num, port_cfg in cfg["ports"].items():
new.configure_port(
router.configure_port(
port=port_num,
ip_address=port_cfg["ip_address"],
subnet_mask=IPv4Address(port_cfg.get("subnet_mask", "255.255.255.0")),
)
if "acl" in cfg:
new.acl._default_config = cfg["acl"] # save the config to allow resetting
new.acl._reset_rules_to_default() # read the config and apply rules
router.acl._default_config = cfg["acl"] # save the config to allow resetting
router.acl._reset_rules_to_default() # read the config and apply rules
if "routes" in cfg:
for route in cfg.get("routes"):
new.route_table.add_route(
router.route_table.add_route(
address=IPv4Address(route.get("address")),
subnet_mask=IPv4Address(route.get("subnet_mask", "255.255.255.0")),
next_hop_ip_address=IPv4Address(route.get("next_hop_ip_address")),
metric=float(route.get("metric", 0)),
)
return new
return router