Skip to content
Snippets Groups Projects
Commit bcda1261 authored by KERDREUX Jerome's avatar KERDREUX Jerome
Browse files

Format (ruff + black config)

parent 0f9e33c2
No related branches found
No related tags found
1 merge request!1First try of type hints
......@@ -29,3 +29,9 @@ dependencies = [
Homepage = "https://recherche.imt-atlantique.fr/xaal/"
Documentation = "https://redmine.imt-atlantique.fr/projects/xaal/repository/xaal/entry/code/Python/branches/0.7/libs/lib/README.rst"
Source = "https://redmine.imt-atlantique.fr/projects/xaal/repository/xaal/show/code/Python/branches/0.7/libs/lib"
[tool.ruff]
line-length = 120
[tool.black]
line-length = 120
......@@ -45,32 +45,33 @@ ALIVE_ADDR = UUID("00000000-0000-0000-0000-000000000000")
class MessageType(Enum):
NOTIFY = 0
NOTIFY = 0
REQUEST = 1
REPLY = 2
REPLY = 2
class MessageAction(Enum):
ALIVE = "alive"
IS_ALIVE = "is_alive"
ALIVE = "alive"
IS_ALIVE = "is_alive"
ATTRIBUTES_CHANGE = "attributes_change"
GET_ATTRIBUTES = "get_attributes"
GET_DESCRIPTION = "get_description"
GET_ATTRIBUTES = "get_attributes"
GET_DESCRIPTION = "get_description"
class Message(object):
"""Message object used for incomming & outgoint message"""
__slots__ = ['version', 'timestamp', 'source', 'dev_type', 'msg_type', 'action', 'body', '__targets']
__slots__ = ["version", "timestamp", "source", "dev_type", "msg_type", "action", "body", "__targets"]
def __init__(self):
self.version = config.STACK_VERSION # message API version
self.__targets = [] # target property
self.timestamp = None # message timestamp
self.source = None # message source
self.dev_type = None # message dev_type
self.msg_type = None # message type
self.action = None # message action
self.body = {} # message body
self.__targets = [] # target property
self.timestamp = None # message timestamp
self.source = None # message source
self.dev_type = None # message dev_type
self.msg_type = None # message type
self.action = None # message action
self.body = {} # message body
@property
def targets(self) -> list:
......@@ -98,12 +99,12 @@ class Message(object):
r.append(["msg_type", MessageType(self.msg_type)])
r.append(["action", self.action])
if self.body:
tmp=""
for k,v in self.body.items():
k = k + ':'
v = pprint.pformat(v,width=55)
tmp = tmp+"- %-12s %s\n" % (k,v)
#tmp = tmp.strip()
tmp = ""
for k, v in self.body.items():
k = k + ":"
v = pprint.pformat(v, width=55)
tmp = tmp + "- %-12s %s\n" % (k, v)
# tmp = tmp.strip()
r.append(["body", tmp])
print(tabulate(r, headers=["Fied", "Value"], tablefmt="psql"))
......@@ -187,8 +188,8 @@ class MessageFactory(object):
buf.append(msg.body)
clear = cbor.dumps(buf)
# Additionnal Data == cbor serialization of the targets array
ad = result[3]
nonce = build_nonce(msg.timestamp)
ad = result[3]
nonce = build_nonce(msg.timestamp)
payload = pysodium.crypto_aead_chacha20poly1305_ietf_encrypt(clear, ad, nonce, self.cipher_key)
# Final CBOR serialization
......@@ -230,7 +231,7 @@ class MessageFactory(object):
# Replay attack, window fixed to CIPHER_WINDOW in seconds
now = build_timestamp()[0] # test done only on seconds ...
if msg_time < (now - config.cipher_window):
raise MessageParserError("Potential replay attack, message too old: %d sec" % round(now - msg_time) )
raise MessageParserError("Potential replay attack, message too old: %d sec" % round(now - msg_time))
if msg_time > (now + config.cipher_window):
raise MessageParserError("Potential replay attack, message too young: %d sec" % round(now - msg_time))
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment