Compare commits
29 Commits
0a442e4ab5
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
| a11629f543 | |||
| 11f96a6069 | |||
| 421cc06d6c | |||
| aef2761f9f | |||
| 607430ca69 | |||
| f05ffdd86d | |||
| ddb5194320 | |||
| 003b2ff367 | |||
| a7178b6f86 | |||
| ee5bc790ea | |||
| f8c215d380 | |||
| 0df5286c0e | |||
| 000a00fec8 | |||
| 405f71f621 | |||
| 82378e0bd7 | |||
| e85fa8ff29 | |||
| 650dd842ce | |||
| 1d60043f37 | |||
| 8849b8488c | |||
| ac9ae74a0b | |||
| 501be198c4 | |||
| 4796d1587b | |||
| deba76228b | |||
| 428c4b8661 | |||
| 93e0676147 | |||
| 2d8a1897f0 | |||
| e3f9fbf36b | |||
| 36187c18b3 | |||
| 2eb04e0428 |
1
.gitignore
vendored
1
.gitignore
vendored
@@ -1,2 +1,3 @@
|
||||
__pycache__
|
||||
dist/
|
||||
docs/_build
|
||||
|
||||
23
README.md
23
README.md
@@ -2,6 +2,20 @@
|
||||
|
||||
Another attempt for a modern client library to the Mitel OM Application XML Interface.
|
||||
|
||||
## Install
|
||||
|
||||
Without any additional dependencies:
|
||||
|
||||
```
|
||||
pip install "mitel_ommclient2 @ git+https://git.clerie.de/clerie/mitel_ommclient2.git@main"
|
||||
```
|
||||
|
||||
Add dependencies to enable secret handling, if you need it.
|
||||
|
||||
```
|
||||
pip install "mitel_ommclient2[crypt] @ git+https://git.clerie.de/clerie/mitel_ommclient2.git@main"
|
||||
```
|
||||
|
||||
## Quicksart
|
||||
|
||||
Just some examples to give you an idea what this does.
|
||||
@@ -23,6 +37,15 @@ r = c.connection.request(m)
|
||||
|
||||
Consult class documentation for more in depth examples and options.
|
||||
|
||||
## Interactive CLI
|
||||
|
||||
The package installs a script called `ommclient2`.
|
||||
This allowes basic interactive testing of the library.
|
||||
|
||||
```
|
||||
ommclient2 --help
|
||||
```
|
||||
|
||||
## Attribution
|
||||
|
||||
This software is inspired by `python-mitel` by Thomas and n-st.
|
||||
|
||||
27
flake.lock
generated
Normal file
27
flake.lock
generated
Normal file
@@ -0,0 +1,27 @@
|
||||
{
|
||||
"nodes": {
|
||||
"nixpkgs": {
|
||||
"locked": {
|
||||
"lastModified": 1665732960,
|
||||
"narHash": "sha256-WBZ+uSHKFyjvd0w4inbm0cNExYTn8lpYFcHEes8tmec=",
|
||||
"owner": "NixOS",
|
||||
"repo": "nixpkgs",
|
||||
"rev": "4428e23312933a196724da2df7ab78eb5e67a88e",
|
||||
"type": "github"
|
||||
},
|
||||
"original": {
|
||||
"owner": "NixOS",
|
||||
"ref": "nixos-unstable",
|
||||
"repo": "nixpkgs",
|
||||
"type": "github"
|
||||
}
|
||||
},
|
||||
"root": {
|
||||
"inputs": {
|
||||
"nixpkgs": "nixpkgs"
|
||||
}
|
||||
}
|
||||
},
|
||||
"root": "root",
|
||||
"version": 7
|
||||
}
|
||||
49
flake.nix
Normal file
49
flake.nix
Normal file
@@ -0,0 +1,49 @@
|
||||
{
|
||||
inputs = {
|
||||
nixpkgs.url = "github:NixOS/nixpkgs/nixos-unstable";
|
||||
};
|
||||
outputs = { self, nixpkgs, ... }: {
|
||||
packages.x86_64-linux = let
|
||||
pkgs = import nixpkgs {
|
||||
system = "x86_64-linux";
|
||||
};
|
||||
in {
|
||||
mitel-ommclient2 = pkgs.python3Packages.buildPythonPackage rec {
|
||||
pname = "mitel-ommclient2";
|
||||
version = "0.0.1";
|
||||
|
||||
src = ./.;
|
||||
|
||||
outputs = [
|
||||
"out"
|
||||
"doc"
|
||||
];
|
||||
|
||||
nativeBuildInputs = [
|
||||
pkgs.python3Packages.sphinxHook
|
||||
];
|
||||
|
||||
format = "pyproject";
|
||||
|
||||
buildInputs = [ pkgs.python3Packages.hatchling ];
|
||||
propagatedBuildInputs = [ pkgs.python3Packages.rsa ];
|
||||
|
||||
pythonImportsCheck = [ "mitel_ommclient2" ];
|
||||
};
|
||||
default = self.packages.x86_64-linux.mitel-ommclient2;
|
||||
};
|
||||
|
||||
apps.x86_64-linux = {
|
||||
ommclient2 = {
|
||||
type = "app";
|
||||
program = self.packages.x86_64-linux.mitel-ommclient2 + "/bin/ommclient2";
|
||||
};
|
||||
default = self.apps.x86_64-linux.ommclient2;
|
||||
};
|
||||
|
||||
hydraJobs = {
|
||||
inherit (self)
|
||||
packages;
|
||||
};
|
||||
};
|
||||
}
|
||||
189
mitel_ommclient2/cli.py
Executable file
189
mitel_ommclient2/cli.py
Executable file
@@ -0,0 +1,189 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
import argparse
|
||||
import getpass
|
||||
import time
|
||||
import traceback
|
||||
|
||||
from . import OMMClient2
|
||||
from .exceptions import EAuth, ENoEnt
|
||||
from .messages import GetAccount, Ping
|
||||
|
||||
# exit handling with argparse is a bit broken even with exit_on_error=False, so we hack this
|
||||
def error_instead_exit(self, message):
|
||||
raise argparse.ArgumentError(None, message)
|
||||
argparse.ArgumentParser.error = error_instead_exit
|
||||
|
||||
def format_child_type(t):
|
||||
return " {}\n{}".format(t.__class__.__name__, "\n".join(["{:<30} {}".format(key, value) for key, value in t._attrs.items()]))
|
||||
|
||||
def format_list(v):
|
||||
return "\n\n\n\n".join(format_child_type(d) for d in v)
|
||||
|
||||
return fl
|
||||
|
||||
def main():
|
||||
connect_parser = argparse.ArgumentParser(prog='ommclient2')
|
||||
connect_parser.add_argument("-n", dest="hostname", default="127.0.0.1")
|
||||
connect_parser.add_argument("-u", dest="username", default="omm")
|
||||
connect_parser.add_argument("-p", dest="password")
|
||||
connect_parser.add_argument("--ommsync", dest="ommsync", action='store_true', help="Log in with ommsync mode")
|
||||
connect_parser.add_argument("subcommand", nargs="*")
|
||||
args = connect_parser.parse_args()
|
||||
|
||||
hostname = args.hostname
|
||||
username = args.username
|
||||
password = args.password
|
||||
ommsync = args.ommsync
|
||||
subcommand = args.subcommand
|
||||
|
||||
if not password:
|
||||
password = getpass.getpass(prompt="OMM password for {}@{}:".format(username, hostname))
|
||||
|
||||
try:
|
||||
c = OMMClient2(hostname, username, password, ommsync=ommsync)
|
||||
except EAuth:
|
||||
print("Authentication failed")
|
||||
exit(1)
|
||||
except TimeoutError:
|
||||
print("OMM unreachable")
|
||||
exit(1)
|
||||
|
||||
parser = argparse.ArgumentParser(prog="ommclient2", add_help=False, exit_on_error=False)
|
||||
subparsers = parser.add_subparsers()
|
||||
|
||||
def add_parser(command_name, func, format=None, args={}):
|
||||
subp = subparsers.add_parser(command_name, help=func.__doc__.strip().split("\n")[0], description=func.__doc__)
|
||||
if format is not None:
|
||||
subp.set_defaults(func=func, format=format)
|
||||
else:
|
||||
subp.set_defaults(func=func)
|
||||
|
||||
for a, t in args.items():
|
||||
subp.add_argument(a, type=t)
|
||||
|
||||
return subp
|
||||
|
||||
parser_exit = subparsers.add_parser("exit")
|
||||
parser_exit.set_defaults(func=exit)
|
||||
|
||||
parser_get_account = add_parser("attach_user_device", func=c.attach_user_device, format=format_list, args={
|
||||
"uid": int,
|
||||
"ppn": int,
|
||||
})
|
||||
|
||||
parser_get_account = add_parser("create_user", func=c.create_user, format=format_child_type, args={
|
||||
"num": str,
|
||||
})
|
||||
|
||||
parser_get_account = add_parser("detach_user_device", func=c.detach_user_device, format=format_list, args={
|
||||
"uid": int,
|
||||
"ppn": int,
|
||||
})
|
||||
|
||||
parser_get_account = add_parser("detach_user_device_by_device", func=c.detach_user_device_by_device, format=format_list, args={
|
||||
"ppn": int,
|
||||
})
|
||||
|
||||
parser_get_account = add_parser("detach_user_device_by_user", func=c.detach_user_device_by_user, format=format_list, args={
|
||||
"uid": int,
|
||||
})
|
||||
|
||||
parser_get_account = add_parser("encrypt", func=c.encrypt, args={
|
||||
"secret": str,
|
||||
})
|
||||
|
||||
parser_get_account = add_parser("get_account", func=c.get_account, format=format_child_type, args={
|
||||
"id": int,
|
||||
})
|
||||
|
||||
parser_get_account = add_parser("get_device", func=c.get_device, format=format_child_type, args={
|
||||
"ppn": int,
|
||||
})
|
||||
|
||||
parser_get_account = add_parser("get_devices", func=c.get_devices, format=format_list)
|
||||
|
||||
parser_get_account = add_parser("get_publickey", func=c.get_publickey)
|
||||
|
||||
parser_get_account = add_parser("get_user", func=c.get_user, format=format_child_type, args={
|
||||
"uid": int,
|
||||
})
|
||||
|
||||
parser_get_account = add_parser("get_users", func=c.get_users, format=format_list)
|
||||
|
||||
parser_help = subparsers.add_parser("help")
|
||||
parser_help.set_defaults(func=parser.format_help)
|
||||
|
||||
parser_ping = subparsers.add_parser("ping")
|
||||
parser_ping.set_defaults(func=lambda *args, **kwargs: "pong" if c.ping(*args, **kwargs) else "error")
|
||||
|
||||
parser_get_account = add_parser("set_user_name", func=c.set_user_name, args={
|
||||
"uid": int,
|
||||
"name": str,
|
||||
})
|
||||
|
||||
parser_get_account = add_parser("set_user_num", func=c.set_user_num, args={
|
||||
"uid": int,
|
||||
"num": str,
|
||||
})
|
||||
|
||||
parser_get_account = add_parser("set_user_relation_dynamic", func=c.set_user_relation_dynamic, args={
|
||||
"uid": int,
|
||||
})
|
||||
|
||||
parser_get_account = add_parser("set_user_relation_fixed", func=c.set_user_relation_fixed, args={
|
||||
"uid": int,
|
||||
})
|
||||
|
||||
parser_get_account = add_parser("set_user_sipauth", func=c.set_user_sipauth, args={
|
||||
"uid": int,
|
||||
"sipAuthId": str,
|
||||
"sipPw": str,
|
||||
})
|
||||
|
||||
if subcommand:
|
||||
try:
|
||||
args = parser.parse_args(subcommand)
|
||||
except argparse.ArgumentError as e:
|
||||
print("argument error:", e.message)
|
||||
exit(1)
|
||||
v = dict(vars(args))
|
||||
v.pop("func")
|
||||
format = lambda r: r
|
||||
if v.get("format") is not None:
|
||||
format = v.get("format")
|
||||
v.pop("format")
|
||||
try:
|
||||
r = args.func(**v)
|
||||
except Exception as e:
|
||||
print("".join(traceback.format_exception(type(e), e, e.__traceback__)))
|
||||
exit(1)
|
||||
print(format(r))
|
||||
exit()
|
||||
|
||||
print("OMMClient")
|
||||
parser.print_help()
|
||||
|
||||
while True:
|
||||
i = input("> ").split()
|
||||
try:
|
||||
args = parser.parse_args(i)
|
||||
except argparse.ArgumentError as e:
|
||||
print("argument error:", e.message)
|
||||
continue
|
||||
|
||||
v = dict(vars(args))
|
||||
v.pop("func")
|
||||
format = lambda r: r
|
||||
if v.get("format") is not None:
|
||||
format = v.get("format")
|
||||
v.pop("format")
|
||||
try:
|
||||
r = args.func(**v)
|
||||
except Exception as e:
|
||||
print("".join(traceback.format_exception(type(e), e, e.__traceback__)))
|
||||
continue
|
||||
print(format(r))
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,8 +1,16 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
import base64
|
||||
try:
|
||||
# This is is only dependency not from the modules inlcuded in python by default, so we make it optional
|
||||
import rsa
|
||||
except ImportError:
|
||||
rsa = None
|
||||
|
||||
from .connection import Connection
|
||||
from . import exceptions
|
||||
from . import messages
|
||||
from . import types
|
||||
|
||||
class OMMClient2:
|
||||
"""
|
||||
@@ -47,10 +55,153 @@ class OMMClient2:
|
||||
m = messages.Open()
|
||||
m.username = self._username
|
||||
m.password = self._password
|
||||
m.UserDeviceSyncClient = self._ommsync
|
||||
if self._ommsync:
|
||||
m.UserDeviceSyncClient = "true"
|
||||
r = self.connection.request(m)
|
||||
r.raise_on_error()
|
||||
|
||||
def attach_user_device(self, uid, ppn):
|
||||
"""
|
||||
Attach user to device
|
||||
|
||||
:param uid: User id
|
||||
:param ppn: Device id
|
||||
|
||||
Requires ommsync=True
|
||||
"""
|
||||
t_u = types.PPUserType()
|
||||
t_u.uid = uid
|
||||
t_u.ppn = ppn
|
||||
t_u.relType = types.PPRelTypeType("Dynamic")
|
||||
t_d = types.PPDevType()
|
||||
t_d.ppn = ppn
|
||||
t_d.uid = uid
|
||||
t_d.relType = types.PPRelTypeType("Dynamic")
|
||||
m = messages.SetPP()
|
||||
m.childs.user = [t_u]
|
||||
m.childs.pp = [t_d]
|
||||
r = self.connection.request(m)
|
||||
r.raise_on_error()
|
||||
if r.childs.user is None:
|
||||
return None
|
||||
return r.childs.user[0], r.childs.pp[0]
|
||||
|
||||
def create_user(self, num):
|
||||
"""
|
||||
Create PP user
|
||||
|
||||
:param num: User number
|
||||
"""
|
||||
t = types.PPUserType()
|
||||
t.num = num
|
||||
m = messages.CreatePPUser()
|
||||
m.childs.user = [t]
|
||||
r = self.connection.request(m)
|
||||
r.raise_on_error()
|
||||
if r.childs.user is None:
|
||||
return None
|
||||
return r.childs.user[0]
|
||||
|
||||
def detach_user_device(self, uid, ppn):
|
||||
"""
|
||||
Detach user from device
|
||||
|
||||
:param uid: User id
|
||||
:param ppn: Device id
|
||||
|
||||
Requires ommsync=True
|
||||
"""
|
||||
t_u = types.PPUserType()
|
||||
t_u.uid = uid
|
||||
t_u.ppn = 0
|
||||
t_u.relType = types.PPRelTypeType("Unbound")
|
||||
t_d = types.PPDevType()
|
||||
t_d.ppn = ppn
|
||||
t_d.uid = 0
|
||||
t_d.relType = types.PPRelTypeType("Unbound")
|
||||
m = messages.SetPP()
|
||||
m.childs.user = [t_u]
|
||||
m.childs.pp = [t_d]
|
||||
r = self.connection.request(m)
|
||||
r.raise_on_error()
|
||||
if r.childs.user is None:
|
||||
return None
|
||||
return r.childs.user[0], r.childs.pp[0]
|
||||
|
||||
def detach_user_device_by_user(self, uid):
|
||||
"""
|
||||
Detach user from device
|
||||
|
||||
This just requires the user id
|
||||
|
||||
:param uid: User id
|
||||
|
||||
Requires ommsync=True
|
||||
"""
|
||||
u = self.get_user(uid)
|
||||
return self.detach_user_device(uid, u.ppn)
|
||||
|
||||
def detach_user_device_by_device(self, ppn):
|
||||
"""
|
||||
Detach user from device
|
||||
|
||||
This just requires the device id
|
||||
|
||||
:param ppn: Device id
|
||||
|
||||
Requires ommsync=True
|
||||
"""
|
||||
d = self.get_device(ppn)
|
||||
return self.detach_user_device(d.uid, ppn)
|
||||
|
||||
def encrypt(self, secret):
|
||||
"""
|
||||
Encrypt secret for OMM
|
||||
|
||||
Required rsa module to be installed
|
||||
|
||||
:param secret: String to encrypt
|
||||
"""
|
||||
|
||||
if rsa is None:
|
||||
raise Exception("rsa module is required for excryption")
|
||||
publickey = self.get_publickey()
|
||||
pubkey = rsa.PublicKey(*publickey)
|
||||
byte_secret = secret.encode('utf8')
|
||||
byte_encrypt = rsa.encrypt(byte_secret, pubkey)
|
||||
encrypt = base64.b64encode(byte_encrypt).decode("utf8")
|
||||
return encrypt
|
||||
|
||||
def find_devices(self, filter):
|
||||
"""
|
||||
Get all devices matching a filter
|
||||
|
||||
:param filter: function taking one parameter which is a device, returns True to keep, False to discard
|
||||
|
||||
Usage::
|
||||
|
||||
>>> c.find_devices(lambda d: d.relType == mitel_ommclient2.types.PPRelTypeType("Unbound"))
|
||||
"""
|
||||
|
||||
for d in self.get_devices():
|
||||
if filter(d):
|
||||
yield d
|
||||
|
||||
def find_users(self, filter):
|
||||
"""
|
||||
Get all users matching a filter
|
||||
|
||||
:param filter: function taking one parameter which is a user, returns True to keep, False to discard
|
||||
|
||||
Usage::
|
||||
|
||||
>>> c.find_users(lambda u: u.num.startswith("9998"))
|
||||
"""
|
||||
|
||||
for u in self.get_users():
|
||||
if filter(u):
|
||||
yield u
|
||||
|
||||
def get_account(self, id):
|
||||
"""
|
||||
Get account
|
||||
@@ -104,6 +255,15 @@ class OMMClient2:
|
||||
# Determine next possible ppn
|
||||
next_ppn = int(pp.ppn) + 1
|
||||
|
||||
def get_publickey(self):
|
||||
"""
|
||||
Get public key for encrypted values
|
||||
"""
|
||||
m = messages.GetPublicKey()
|
||||
r = self.connection.request(m)
|
||||
r.raise_on_error()
|
||||
return int(r.modulus, 16), int(r.exponent, 16)
|
||||
|
||||
def get_user(self, uid):
|
||||
"""
|
||||
Get PP user
|
||||
@@ -152,3 +312,83 @@ class OMMClient2:
|
||||
if r.errCode is None:
|
||||
return True
|
||||
return False
|
||||
|
||||
def set_user_name(self, uid, name):
|
||||
"""
|
||||
Set PP user name
|
||||
|
||||
:param uid: User id
|
||||
:param name: User name
|
||||
"""
|
||||
t = types.PPUserType()
|
||||
t.uid = uid
|
||||
t.name = name
|
||||
m = messages.SetPPUser()
|
||||
m.childs.user = [t]
|
||||
r = self.connection.request(m)
|
||||
r.raise_on_error()
|
||||
if r.childs.user is None:
|
||||
return None
|
||||
return r.childs.user[0]
|
||||
|
||||
def set_user_num(self, uid, num):
|
||||
"""
|
||||
Set PP user number
|
||||
|
||||
:param uid: User id
|
||||
:param num: User number
|
||||
"""
|
||||
t = types.PPUserType()
|
||||
t.uid = uid
|
||||
t.num = num
|
||||
m = messages.SetPPUser()
|
||||
m.childs.user = [t]
|
||||
r = self.connection.request(m)
|
||||
r.raise_on_error()
|
||||
if r.childs.user is None:
|
||||
return None
|
||||
return r.childs.user[0]
|
||||
|
||||
def set_user_relation_dynamic(self, uid):
|
||||
"""
|
||||
Set PP user to PP device relation to dynamic type
|
||||
|
||||
:param uid: User id
|
||||
"""
|
||||
m = messages.SetPPUserDevRelation()
|
||||
m.uid = uid
|
||||
m.relType = types.PPRelTypeType("Dynamic")
|
||||
r = self.connection.request(m)
|
||||
r.raise_on_error()
|
||||
|
||||
def set_user_relation_fixed(self, uid):
|
||||
"""
|
||||
Set PP user to PP device relation to fixed type
|
||||
|
||||
:param uid: User id
|
||||
"""
|
||||
m = messages.SetPPUserDevRelation()
|
||||
m.uid = uid
|
||||
m.relType = types.PPRelTypeType("Fixed")
|
||||
r = self.connection.request(m)
|
||||
r.raise_on_error()
|
||||
|
||||
def set_user_sipauth(self, uid, sipAuthId, sipPw):
|
||||
"""
|
||||
Set PP user sip credentials
|
||||
|
||||
:param uid: User id
|
||||
:param sipAuthId: SIP user name
|
||||
:param sipPw: Plain text password
|
||||
"""
|
||||
t = types.PPUserType()
|
||||
t.uid = uid
|
||||
t.sipAuthId = sipAuthId
|
||||
t.sipPw = self.encrypt(sipPw)
|
||||
m = messages.SetPPUser()
|
||||
m.childs.user = [t]
|
||||
r = self.connection.request(m)
|
||||
r.raise_on_error()
|
||||
if r.childs.user is None:
|
||||
return None
|
||||
return r.childs.user[0]
|
||||
|
||||
@@ -26,6 +26,7 @@ class Connection:
|
||||
self._host = host
|
||||
self._port = port
|
||||
self._socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
self._socket.settimeout(2)
|
||||
|
||||
self._seq = 0 # state of the sequence number generator
|
||||
self._requests = {} # waiting area for pending responses
|
||||
@@ -55,8 +56,11 @@ class Connection:
|
||||
if select.select([self._socket], [], []) != ([], [], []):
|
||||
# wait for data availiable
|
||||
while True:
|
||||
# fill buffer with one message
|
||||
data = self._socket.recv(1024)
|
||||
try:
|
||||
# fill buffer with one message
|
||||
data = self._socket.recv(1024)
|
||||
except BlockingIOError:
|
||||
continue
|
||||
|
||||
if not data:
|
||||
# buffer is empty
|
||||
|
||||
@@ -44,8 +44,11 @@ class Message:
|
||||
|
||||
def __setattr__(self, name, value):
|
||||
if name in self.CHILDS.keys():
|
||||
if self.CHILDS[name] is not None and type(value) != self.CHILDS[name]:
|
||||
if not isinstance(value, list):
|
||||
raise TypeError()
|
||||
for v in value:
|
||||
if self.CHILDS[name] is not None and type(v) != self.CHILDS[name]:
|
||||
raise TypeError()
|
||||
self._child_dict[name] = value
|
||||
else:
|
||||
object.__setattr__(self, name, value)
|
||||
@@ -55,8 +58,8 @@ class Message:
|
||||
self.name = name
|
||||
if not self.name:
|
||||
self.name = self.__class__.__name__
|
||||
self._attrs = attrs
|
||||
self._childs = childs
|
||||
self._attrs = {} | attrs
|
||||
self._childs = {} | childs
|
||||
self.childs = self.Childs(self.CHILDS, self._childs)
|
||||
|
||||
def __getattr__(self, name):
|
||||
@@ -76,7 +79,7 @@ class Message:
|
||||
object.__setattr__(self, name, value)
|
||||
|
||||
def __repr__(self):
|
||||
return "{}({}, {}, {})".format(self.__class__.__name__, self.name, repr(self._attrs), repr(self._childs))
|
||||
return "{}({}, {}, {})".format(self.__class__.__name__, repr(self.name), repr(self._attrs), repr(self._childs))
|
||||
|
||||
|
||||
class Request(Message):
|
||||
@@ -131,11 +134,16 @@ def response_type(c):
|
||||
RESPONSE_TYPES[c.__name__] = c
|
||||
return c
|
||||
|
||||
from .createppuser import CreatePPUser, CreatePPUserResp
|
||||
from .getaccount import GetAccount, GetAccountResp
|
||||
from .getppdev import GetPPDev, GetPPDevResp
|
||||
from .getppuser import GetPPUser, GetPPUserResp
|
||||
from .getpublickey import GetPublicKey, GetPublicKeyResp
|
||||
from .open import Open, OpenResp
|
||||
from .ping import Ping, PingResp
|
||||
from .setpp import SetPP, SetPPResp
|
||||
from .setppuser import SetPPUser, SetPPUserResp
|
||||
from .setppuserdevrelation import SetPPUserDevRelation, SetPPUserDevRelationResp
|
||||
|
||||
def construct(request):
|
||||
"""
|
||||
@@ -149,12 +157,13 @@ def construct(request):
|
||||
root.setAttribute(str(k), str(v))
|
||||
|
||||
|
||||
for k, v in request._childs.items():
|
||||
child = message.createElement(k)
|
||||
if v is not None:
|
||||
for c_k, c_v in v.items():
|
||||
child.setAttribute(str(c_k), str(c_v))
|
||||
root.appendChild(child)
|
||||
for child_name, child_list in request._childs.items():
|
||||
if child_list is not None:
|
||||
for child_list_item in child_list:
|
||||
child = message.createElement(child_name)
|
||||
for child_item_key, child_item_value in child_list_item._attrs.items():
|
||||
child.setAttribute(str(child_item_key), str(child_item_value))
|
||||
root.appendChild(child)
|
||||
return root.toxml()
|
||||
|
||||
def parse(message):
|
||||
|
||||
18
mitel_ommclient2/messages/createppuser.py
Normal file
18
mitel_ommclient2/messages/createppuser.py
Normal file
@@ -0,0 +1,18 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from . import Request, Response, request_type, response_type
|
||||
from ..types import PPUserType
|
||||
|
||||
|
||||
@request_type
|
||||
class CreatePPUser(Request):
|
||||
CHILDS = {
|
||||
"user": PPUserType,
|
||||
}
|
||||
|
||||
|
||||
@response_type
|
||||
class CreatePPUserResp(Response):
|
||||
CHILDS = {
|
||||
"user": PPUserType,
|
||||
}
|
||||
16
mitel_ommclient2/messages/getpublickey.py
Normal file
16
mitel_ommclient2/messages/getpublickey.py
Normal file
@@ -0,0 +1,16 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from . import Request, Response, request_type, response_type
|
||||
|
||||
|
||||
@request_type
|
||||
class GetPublicKey(Request):
|
||||
pass
|
||||
|
||||
|
||||
@response_type
|
||||
class GetPublicKeyResp(Response):
|
||||
FIELDS = {
|
||||
"modulus": str,
|
||||
"exponent": str,
|
||||
}
|
||||
20
mitel_ommclient2/messages/setpp.py
Normal file
20
mitel_ommclient2/messages/setpp.py
Normal file
@@ -0,0 +1,20 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from . import Request, Response, request_type, response_type
|
||||
from ..types import PPDevType, PPUserType
|
||||
|
||||
|
||||
@request_type
|
||||
class SetPP(Request):
|
||||
CHILDS = {
|
||||
"pp": PPDevType,
|
||||
"user": PPUserType,
|
||||
}
|
||||
|
||||
|
||||
@response_type
|
||||
class SetPPResp(Response):
|
||||
CHILDS = {
|
||||
"pp": PPDevType,
|
||||
"user": PPUserType,
|
||||
}
|
||||
18
mitel_ommclient2/messages/setppuser.py
Normal file
18
mitel_ommclient2/messages/setppuser.py
Normal file
@@ -0,0 +1,18 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from . import Request, Response, request_type, response_type
|
||||
from ..types import PPUserType
|
||||
|
||||
|
||||
@request_type
|
||||
class SetPPUser(Request):
|
||||
CHILDS = {
|
||||
"user": PPUserType,
|
||||
}
|
||||
|
||||
|
||||
@response_type
|
||||
class SetPPUserResp(Response):
|
||||
CHILDS = {
|
||||
"user": PPUserType,
|
||||
}
|
||||
20
mitel_ommclient2/messages/setppuserdevrelation.py
Normal file
20
mitel_ommclient2/messages/setppuserdevrelation.py
Normal file
@@ -0,0 +1,20 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from . import Request, Response, request_type, response_type
|
||||
from ..types import PPRelTypeType
|
||||
|
||||
|
||||
@request_type
|
||||
class SetPPUserDevRelation(Request):
|
||||
FIELDS = {
|
||||
"uid": int,
|
||||
"relType": PPRelTypeType,
|
||||
}
|
||||
|
||||
|
||||
@response_type
|
||||
class SetPPUserDevRelationResp(Response):
|
||||
FIELDS = {
|
||||
"uid": int,
|
||||
"relType": PPRelTypeType,
|
||||
}
|
||||
@@ -39,12 +39,16 @@ class ChildType:
|
||||
return "{}({})".format(self.__class__.__name__, repr(self._attrs))
|
||||
|
||||
def cast_dict_to_childtype(t, d):
|
||||
errors = {} # collect unknown keys
|
||||
for k, v in d.items():
|
||||
if k in t.FIELDS.keys():
|
||||
if t.FIELDS[k] is not None and type(v) != t.FIELDS[k]:
|
||||
d[k] = t.FIELDS[k](v)
|
||||
else:
|
||||
raise KeyError()
|
||||
errors[k] = v
|
||||
|
||||
if errors != {}:
|
||||
raise KeyError("The following keys are unknown for '{}': {}".format(t.__name__, errors))
|
||||
|
||||
return t(d)
|
||||
|
||||
@@ -68,6 +72,9 @@ class EnumType:
|
||||
def __repr__(self):
|
||||
return "{}({})".format(self.__class__.__name__, repr(self.value))
|
||||
|
||||
def __eq__(self, other):
|
||||
return isinstance(other, type(self)) and self.value == other.value
|
||||
|
||||
|
||||
class CallForwardStateType(EnumType):
|
||||
VALUES = [
|
||||
@@ -92,7 +99,11 @@ class MonitoringStateType(EnumType):
|
||||
|
||||
|
||||
class PPRelTypeType(EnumType):
|
||||
VALUES = None
|
||||
VALUES = [
|
||||
"Fixed",
|
||||
"Dynamic",
|
||||
"Unbound",
|
||||
]
|
||||
|
||||
|
||||
class AccountType(ChildType):
|
||||
@@ -130,6 +141,18 @@ class PPDevType(ChildType):
|
||||
"ppDefaultProfileLoaded": bool,
|
||||
"subscribeToPARIOnly": bool,
|
||||
# undocumented
|
||||
"ommId": str,
|
||||
"ommIdAck": str,
|
||||
"timeStampAdmin": int,
|
||||
"timeStampRelation": int,
|
||||
"timeStampRoaming": int,
|
||||
"timeStampSubscription": int,
|
||||
"autoCreate": bool,
|
||||
"roaming": None, # value: 'RoamingComplete'
|
||||
"modicType": str, # value: '01'
|
||||
"locationData": str, # value: '000001000000'
|
||||
"dectIeFixedId": str,
|
||||
"subscriptionId": str,
|
||||
"ppnSec": int,
|
||||
}
|
||||
|
||||
@@ -209,4 +232,7 @@ class PPUserType(ChildType):
|
||||
"keyLockEnable": None,
|
||||
"keyLockPin": None,
|
||||
"keyLockTime": None,
|
||||
"ppnOld": int,
|
||||
"timeStampAdmin": int,
|
||||
"timeStampRelation": int,
|
||||
}
|
||||
|
||||
30
pyproject.toml
Normal file
30
pyproject.toml
Normal file
@@ -0,0 +1,30 @@
|
||||
[build-system]
|
||||
requires = ["hatchling"]
|
||||
build-backend = "hatchling.build"
|
||||
|
||||
[project]
|
||||
name = "mitel_ommclient2"
|
||||
version = "0.0.1"
|
||||
authors = [
|
||||
{ name="clerie", email="hallo@clerie.de" },
|
||||
]
|
||||
description = "Another attempt for a modern client library to the Mitel OM Application XML Interface."
|
||||
readme = "README.md"
|
||||
license = { file="LICENSE" }
|
||||
requires-python = ">=3.7"
|
||||
classifiers = [
|
||||
"Programming Language :: Python :: 3",
|
||||
"License :: OSI Approved :: MIT License",
|
||||
"Operating System :: OS Independent",
|
||||
]
|
||||
|
||||
[project.optional-dependencies]
|
||||
crypt = [
|
||||
"rsa",
|
||||
]
|
||||
|
||||
[project.scripts]
|
||||
ommclient2 = "mitel_ommclient2.cli:main"
|
||||
|
||||
[project.urls]
|
||||
"Source" = "https://git.clerie.de/clerie/mitel_ommclient2"
|
||||
Reference in New Issue
Block a user