gns3api - Simple python module to access the GNS3 API

gns3api.py 6.4KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204
  1. """
  2. Access GNS3 controller via API
  3. """
  4. import os
  5. import sys
  6. import ssl
  7. import json
  8. from base64 import b64encode
  9. try:
  10. import configparser
  11. except ImportError: # fallback to Python 2 module
  12. import ConfigParser as configparser
  13. try:
  14. import http.client as http_client
  15. GNS3BaseException = OSError
  16. except ImportError: # fallback to Python 2 module
  17. import httplib as http_client
  18. GNS3BaseException = IOError
  19. class GNS3ApiException(GNS3BaseException):
  20. """
  21. GNS3 API Exceptions, base class
  22. """
  23. pass
  24. class HTTPClientError(GNS3ApiException):
  25. """
  26. HTTP client library error
  27. """
  28. def __str__(self):
  29. if not self.args[1]:
  30. return '{}'.format(self.args[0])
  31. return '{}: {}'.format(self.args[0], self.args[1])
  32. class HTTPError(GNS3ApiException):
  33. """
  34. HTTP response error
  35. """
  36. def __str__(self):
  37. return '[Status {}] {}'.format(self.args[0], self.args[1])
  38. class GNS3Api:
  39. """
  40. GNS3 API - an API to GNS3
  41. """
  42. def __init__(self, proto='http', host=None, port=3080,
  43. user=None, password=None, verify=True):
  44. """
  45. GNS3 API
  46. :param proto: Protocol (http/https), default 'http'
  47. :param host: Host name or IP, if None the connection parameters
  48. are read from the GNS3 configuration file
  49. :param port; Port number, default 3080
  50. :param user: User name, None for no authentification
  51. :param password: Password
  52. :param verify: Verify CERT (on https), default True
  53. False: no CERT verification
  54. True: verification using the system CA certificates
  55. file: verification using the file and the system CA
  56. """
  57. if host is None or host == '':
  58. (proto, host, port, user, password) = GNS3Api.get_controller_params()
  59. self.controller = "{}://{}:{}".format(proto, host, port)
  60. self.status_code = None
  61. # authentication
  62. self._auth = {}
  63. if user is not None and user != '':
  64. if password is None:
  65. password = ''
  66. self._auth['Authorization'] = 'Basic ' + \
  67. b64encode((user+':'+password).encode('utf-8')).decode('ascii')
  68. # open connection
  69. try:
  70. if proto == 'http':
  71. self._conn = http_client.HTTPConnection(host, port, timeout=10)
  72. elif proto == 'https':
  73. context = ssl.create_default_context()
  74. if isinstance(verify, str):
  75. context.check_hostname = False
  76. context.load_verify_locations(cafile=verify)
  77. elif not verify:
  78. context.check_hostname = False
  79. context.verify_mode = ssl.CERT_NONE
  80. self._conn = http_client.HTTPSConnection(host, port, timeout=10,
  81. context=context)
  82. else:
  83. raise HTTPClientError("UnknownProtocol", proto)
  84. self._conn.connect()
  85. except http_client.HTTPException as err:
  86. raise HTTPClientError(type(err).__name__, str(err))
  87. @staticmethod
  88. def get_controller_params():
  89. """
  90. Returns GNS3 controller connection parameters
  91. :returns: Tuple of protocol, host, port, user, password
  92. """
  93. # find config file
  94. if sys.platform.startswith('win'):
  95. fn_conf = os.path.join(os.path.expandvars('%APPDATA%'), 'GNS3',
  96. 'gns3_server.ini')
  97. else:
  98. fn_conf = os.path.join(os.path.expanduser('~'), '.config', 'GNS3',
  99. 'gns3_server.conf')
  100. # parse config
  101. config = configparser.ConfigParser()
  102. try:
  103. config.read(fn_conf)
  104. serv_conf = dict(config.items('Server'))
  105. except (IOError, OSError, configparser.Error):
  106. serv_conf = {}
  107. # extract config variables
  108. proto = serv_conf.get('protocol', 'http')
  109. host = serv_conf.get('host', '127.0.0.1')
  110. port = int(serv_conf.get('port', 3080))
  111. user = serv_conf.get('user', None)
  112. password = serv_conf.get('password', None)
  113. return (proto, host, port, user, password)
  114. def request(self, method, path, args=None, timeout=60):
  115. """
  116. API request
  117. :param method: HTTP method ('GET'/'PUT'/'POST'/'DELETE')
  118. :param path: URL path, can be a list or tuple
  119. :param args: arguments to the API endpoint
  120. :param timeout: timeout, default 60
  121. :returns: result
  122. """
  123. # json encode args
  124. if args is None:
  125. body = None
  126. else:
  127. body = json.dumps(args, separators=(',', ':'))
  128. # methods are upper case
  129. method.upper()
  130. # make path variable to an URL path
  131. if isinstance(path, (list, tuple)):
  132. path = "/".join(str(x) for x in path)
  133. else:
  134. path = str(path)
  135. if not path.startswith("/"):
  136. path = "/" + path
  137. # send request
  138. if self._conn.timeout != timeout:
  139. self._conn.timeout = timeout
  140. if self._conn.sock:
  141. self._conn.sock.settimeout(timeout)
  142. headers = {'Content-Type': 'application/json',
  143. 'User-Agent': 'GNS3Api'}
  144. headers.update(self._auth)
  145. try:
  146. # send request / get response
  147. self._conn.request(method, path, body, headers=headers)
  148. resp = self._conn.getresponse()
  149. data = resp.read()
  150. if resp.getheader('Content-Type') == 'application/json':
  151. result = json.loads(data.decode('utf-8', errors='ignore'))
  152. else:
  153. result = data
  154. except http_client.HTTPException as err:
  155. raise HTTPClientError(type(err).__name__, str(err))
  156. # check for errors
  157. self.status_code = resp.status
  158. if self.status_code < 200 or self.status_code >= 300:
  159. try:
  160. message = result['message']
  161. except (TypeError, KeyError):
  162. if data is not None and data != b'':
  163. message = data.decode('utf-8', errors='ignore')
  164. else:
  165. message = resp.reason
  166. raise HTTPError(self.status_code, message)
  167. return result
  168. def close(self):
  169. """
  170. Closes HTTP(S) connection
  171. """
  172. self._conn.close()