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

#2041: pre-commit changes.

Browse Source
This commit is contained in:
Nick Todd
2023-11-15 10:57:56 +00:00
parent 195e8a4e84
commit d31fce202c
2 changed files with 12 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/primaite/simulator/network/protocols/ntp.py
View File

@@ -1,10 +1,12 @@
from __future__ import annotations from __future__ import annotations
from datetime import datetime
from ipaddress import IPv4Address from ipaddress import IPv4Address
from typing import Optional from typing import Optional
from pydantic import BaseModel from pydantic import BaseModel
from primaite.simulator.network.protocols.packet import DataPacket from primaite.simulator.network.protocols.packet import DataPacket
from datetime import datetime
class NTPRequest(BaseModel): class NTPRequest(BaseModel):
@@ -33,7 +35,7 @@ class NTPPacket(DataPacket):
ntp_reply: Optional[NTPReply] = None ntp_reply: Optional[NTPReply] = None
def generate_reply(self, time: datetime) -> NTPPacket: def generate_reply(self, time: datetime) -> NTPPacket:
""" Generate a NTPPacket containing the time in a NTPReply object """Generate a NTPPacket containing the time in a NTPReply object.
:param time: datetime object representing the time from the NTP server. :param time: datetime object representing the time from the NTP server.
:return: A new NTPPacket object. :return: A new NTPPacket object.

11
src/primaite/simulator/system/services/ntp/ntp_server.py
View File

@@ -1,4 +1,4 @@
from ipaddress import IPv4Address from datetime import datetime
from typing import Any, Dict, Optional from typing import Any, Dict, Optional
from primaite import getLogger from primaite import getLogger
@@ -6,13 +6,12 @@ from primaite.simulator.network.protocols.ntp import NTPPacket
from primaite.simulator.network.transmission.network_layer import IPProtocol from primaite.simulator.network.transmission.network_layer import IPProtocol
from primaite.simulator.network.transmission.transport_layer import Port from primaite.simulator.network.transmission.transport_layer import Port
from primaite.simulator.system.services.service import Service from primaite.simulator.system.services.service import Service
from datetime import datetime
_LOGGER = getLogger(__name__) _LOGGER = getLogger(__name__)
class NTPServer(Service): class NTPServer(Service):
"""Represents a NTP server as a service""" """Represents a NTP server as a service."""
def __init__(self, **kwargs): def __init__(self, **kwargs):
kwargs["name"] = "NTPServer" kwargs["name"] = "NTPServer"
@@ -60,15 +59,13 @@ class NTPServer(Service):
:return: True if valid NTP request else False. :return: True if valid NTP request else False.
""" """
if not (isinstance(payload, NTPPacket) and if not (isinstance(payload, NTPPacket) and payload.ntp_request.ntp_client):
payload.ntp_request.ntp_client):
_LOGGER.debug(f"{payload} is not a NTPPacket") _LOGGER.debug(f"{payload} is not a NTPPacket")
return False return False
payload: NTPPacket = payload payload: NTPPacket = payload
if payload.ntp_request.ntp_client: if payload.ntp_request.ntp_client:
self.sys_log.info( self.sys_log.info(
f"{self.name}: Received request for {payload.ntp_request.ntp_client} " f"{self.name}: Received request for {payload.ntp_request.ntp_client} " f"from session {session_id}"
f"from session {session_id}"
) )
# generate a reply with the current time # generate a reply with the current time
time = datetime.now() time = datetime.now()