105 lines
2.8 KiB
Python
105 lines
2.8 KiB
Python
import subprocess
|
|
import typing
|
|
|
|
from source import managers
|
|
|
|
|
|
class ChronyManager:
|
|
"""
|
|
Manager for chrony to manipulate time synchronisation
|
|
NOTE: the ideal method would be to use `chronyd.sock` to communicate, but it seems to only work with raw binary data
|
|
"""
|
|
|
|
def __init__(self, manager: "managers.Manager"):
|
|
self.manager = manager
|
|
|
|
self.clear_sources()
|
|
self.deny_client("all")
|
|
|
|
@staticmethod
|
|
def run_command(*args: str) -> str:
|
|
# run the command through the chrony command line tool
|
|
# NOTE(Faraphel): do not check the errors because some codes like 511 can happen
|
|
# if we add a source that already exist
|
|
process = subprocess.run(
|
|
["chronyc", *args],
|
|
stdout=subprocess.PIPE,
|
|
)
|
|
|
|
# return the stdout of the command
|
|
return process.stdout.decode()
|
|
|
|
def get_sources(self) -> typing.Iterator[str]:
|
|
"""
|
|
Get the list of NTP sources
|
|
:return: the list of NTP sources
|
|
"""
|
|
|
|
table: str = self.run_command("sources")
|
|
sources_info = table.strip().split("\n")[2:]
|
|
print(sources_info)
|
|
for source_info in sources_info:
|
|
yield source_info.split(" ")[1]
|
|
|
|
def add_source(self, address: str) -> None:
|
|
"""
|
|
Add a source to chrony
|
|
"""
|
|
|
|
self.run_command("add", "server", address, "iburst")
|
|
|
|
def remove_source(self, address: str) -> None:
|
|
"""
|
|
Remove a source from chrony
|
|
"""
|
|
|
|
self.run_command("delete", address)
|
|
|
|
def clear_sources(self) -> None:
|
|
"""
|
|
Clear the list of sources used by chrony
|
|
"""
|
|
|
|
for source in self.get_sources():
|
|
self.remove_source(source)
|
|
|
|
def get_clients(self) -> typing.Iterator[str]:
|
|
"""
|
|
Get the list of clients using us as a server
|
|
:return: the list of clients
|
|
"""
|
|
|
|
table: str = self.run_command("clients")
|
|
clients_info = table.strip().split("\n")[2:]
|
|
for client_info in clients_info:
|
|
yield client_info.split(" ")[0]
|
|
|
|
def allow_client(self, address: str) -> None:
|
|
"""
|
|
Allow a client to use us as a server
|
|
:param address: the client address
|
|
"""
|
|
|
|
self.run_command("allow", address)
|
|
|
|
def allow_all_clients(self) -> None:
|
|
"""
|
|
Allow all clients to use us as a time-server
|
|
"""
|
|
|
|
self.allow_client("all")
|
|
|
|
def deny_client(self, address: str) -> None:
|
|
"""
|
|
Deny a client to use us as a server
|
|
:param address: the client address
|
|
"""
|
|
|
|
self.run_command("deny", address)
|
|
|
|
def deny_all_clients(self) -> None:
|
|
"""
|
|
Deny all clients from using us as a time-server
|
|
"""
|
|
|
|
self.deny_client("all")
|