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 7.6KB

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