gns3api - Simple python module to access the GNS3 API
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

gns3api.py 6.7KB

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