File: apicli.py

package info (click to toggle)
librouteros 3.1.0-1
  • links: PTS, VCS
  • area: main
  • in suites: bullseye, sid
  • size: 336 kB
  • sloc: python: 1,110; makefile: 127; sh: 8
file content (75 lines) | stat: -rwxr-xr-x 1,983 bytes parent folder | download | duplicates (2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
#!/usr/bin/env python3
# -*- coding: UTF-8 -*-

"""Command line interface for debugging purpouses."""


import logging
import getpass
from argparse import ArgumentParser
from sys import stdout, stdin
from select import select
from os import linesep

from librouteros import connect, ConnectionError, TrapError, FatalError

argParser = ArgumentParser(description='mikrotik api cli interface')
argParser.add_argument(
        'host', type=str,
        help="host to with to connect. may be fqdn, ipv4 or ipv6 address")
argParser.add_argument('-u', '--user', type=str, required=True, help="username")
argParser.add_argument(
        '-p', '--port', type=int, default=8728, help="port to connect to (default 8728)")
args = argParser.parse_args()

mainlog = logging.getLogger('librouteros')
console = logging.StreamHandler(stdout)
mainlog.setLevel(logging.DEBUG)
formatter = logging.Formatter(fmt='%(message)s')
console.setFormatter(formatter)
mainlog.addHandler(console)


def selectloop(api):
    snt = []
    while True:
        proto = api.protocol
        sk = proto.transport.sock

        rlist, wlist, errlist = select([sk, stdin], [], [], None)

        if sk in rlist:
            proto.readSentence()

        if stdin in rlist:
            line = stdin.readline()
            line = line.split(linesep)
            line = line[0]
            if line:
                snt.append(line)
            elif not line and snt:
                proto.writeSentence(snt[0], *snt[1:])
                snt = []


def main():
    pw = getpass.getpass()
    try:
        api = connect(args.host, args.user, pw, logger=mainlog)
    except (TrapError, ConnectionError) as err:
        exit(err)
    except KeyboardInterrupt:
        pass
    else:
        try:
            selectloop(api)
        except KeyboardInterrupt:
            pass
        except (ConnectionError, FatalError) as e:
            print(e)
        finally:
            api.close()


if __name__ == '__main__':
    main()