#!/usr/bin/env python # # Copyright (C) 2016 GNS3 Technologies Inc. # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see . import asyncio import io import ipaddress import json import logging import socket import sys from asyncio import timeout as asynctimeout import aiohttp from fastapi import HTTPException from ..controller.controller_error import ( ComputeConflictError, ComputeError, ControllerBadRequestError, ControllerError, ControllerForbiddenError, ControllerNotFoundError, ControllerTimeoutError, ControllerUnauthorizedError, ) from ..utils import parse_version from ..utils.asyncio import async_iterable_to_stream, locking from ..version import __version__, __version_info__ log = logging.getLogger(__name__) class Compute: """ A GNS3 compute. """ def __init__( self, compute_id, controller=None, protocol="http", host="localhost", port=3080, user=None, password=None, name=None, console_host=None, ssl_context=None, auto_connect=False, ): self._http_session = None assert controller is not None log.info("Create compute %s", compute_id) # if compute_id is None: # self._id = str(uuid.uuid4()) # else: self._id = compute_id self.protocol = protocol self._console_host = console_host self.host = host self.port = port self._user = None self._password = None self._connected = False self._notifications = None self._closed = False # Close mean we are destroying the compute node # True when the user asked to disconnect: blocks every reconnection # path (retry backoff, notification stream reconnect, lazy connect) # until connect() is called again self._manual_disconnect = False self._auto_connect = auto_connect self._controller = controller self._set_auth(user, password) self._cpu_usage_percent = 0 self._memory_usage_percent = 0 self._disk_usage_percent = 0 self._last_error = None self._ssl_context = ssl_context self._capabilities = { "version": "", "platform": "", "cpus": 0, "memory": 0, "disk_size": 0, "node_types": [], "ubridge_tc": None, "ubridge_tap": None, "ubridge_iol_tap": None, "ubridge_bridge_tap": None, } self.name = name # Cache of interfaces on remote host self._interfaces_cache = None # Cached resolution of self._host — socket.gethostbyname is a blocking # call; resolving it on every host_ip access (several times per link # via get_ip_on_same_subnet) freezes the event loop for all coroutines. self._host_ip_cache = None self._connection_failure = 0 def _session(self): if self._http_session is None or self._http_session.closed is True: connector = aiohttp.TCPConnector(force_close=True, ssl_context=self._ssl_context) self._http_session = aiohttp.ClientSession(connector=connector) return self._http_session def _set_auth(self, user, password): """ Set authentication parameters """ if user is None or len(user.strip()) == 0: self._user = None self._password = None self._auth = None else: self._user = user.strip() if password: self._password = password try: self._auth = aiohttp.BasicAuth(self._user, self._password.get_secret_value(), "utf-8") except ValueError as e: log.error(str(e)) else: self._password = None self._auth = aiohttp.BasicAuth(self._user, "") def set_last_error(self, msg): """ Set the last error message for this compute. :param msg: message """ self._last_error = msg async def interfaces(self): """ Get the list of network on compute """ if not self._interfaces_cache: response = await self.get("/network/interfaces") self._interfaces_cache = response.json return self._interfaces_cache async def update(self, **kwargs): if "auto_connect" in kwargs: # stored separately from the generic attributes below so that # asdict() reports it (it is persisted in the database, the # controller-side copy is only a mirror for notifications) self._auto_connect = kwargs.pop("auto_connect") for kw in kwargs: if kw not in ("user", "password"): setattr(self, kw, kwargs[kw]) # It's important to set user and password at the same time if "user" in kwargs or "password" in kwargs: self._set_auth(kwargs.get("user", self._user), kwargs.get("password", self._password)) if self._http_session and not self._http_session.closed: await self._http_session.close() self._connected = False self._controller.notification.controller_emit("compute.updated", self.asdict()) self._controller.save() async def close(self): self._connected = False if self._http_session and not self._http_session.closed: await self._http_session.close() try: if self._notifications: await self._notifications except asyncio.CancelledError: pass self._closed = True @property def name(self): """ :returns: Compute name """ return self._name @name.setter def name(self, name): if name is not None: self._name = name else: if self._user: user = self._user # Due to random user generated by 1.4 it's common to have a very long user if len(user) > 14: user = user[:11] + "..." self._name = f"{self._protocol}://{user}@{self._host}:{self._port}" else: self._name = f"{self._protocol}://{self._host}:{self._port}" @property def connected(self): """ :returns: True if compute node is connected """ return self._connected @property def id(self): """ :returns: Compute identifier (string) """ return self._id @property def host(self): """ :returns: Compute host (string) """ return self._host @host.setter def host(self, host): self._host = host self._host_ip_cache = None # invalidate; re-resolve on next access if self._console_host is None: self._console_host = host @property def host_ip(self): """ Return the IP associated to the host """ if self._host_ip_cache is None: try: self._host_ip_cache = socket.gethostbyname(self._host) except socket.gaierror: self._host_ip_cache = "0.0.0.0" return self._host_ip_cache @property def console_host(self): return self._console_host @property def port(self): """ :returns: Compute port (integer) """ return self._port @port.setter def port(self, port): self._port = port @property def protocol(self): """ :returns: Compute protocol (string) """ return self._protocol @protocol.setter def protocol(self, protocol): self._protocol = protocol @property def user(self): return self._user @user.setter def user(self, value): self._set_auth(value, self._password) @property def password(self): return self._password @password.setter def password(self, value): self._set_auth(self._user, value) @property def cpu_usage_percent(self): return self._cpu_usage_percent @property def memory_usage_percent(self): return self._memory_usage_percent @property def disk_usage_percent(self): return self._disk_usage_percent @property def capabilities(self) -> dict: """ Capabilities reported by this compute (fetched from GET /capabilities at connect time). """ return self._capabilities def asdict(self, topology_dump=False): """ :param topology_dump: Filter to keep only properties require for saving on disk """ if topology_dump: return { "compute_id": self._id, "name": self._name, "protocol": self._protocol, "host": self._host, "port": self._port, } return { "compute_id": self._id, "name": self._name, "protocol": self._protocol, "host": self._host, "port": self._port, "user": self._user, "auto_connect": self._auto_connect, "connected": self._connected, "cpu_usage_percent": self._cpu_usage_percent, "memory_usage_percent": self._memory_usage_percent, "disk_usage_percent": self._disk_usage_percent, "capabilities": self._capabilities, "last_error": self._last_error, } async def download_file(self, project, path): """ Read file of a project and download it :param project: A project object :param path: The path of the file in the project :returns: A file stream """ url = self._getUrl(f"/projects/{project.id}/files/{path}") response = await self._session().request("GET", url, auth=self._auth) if response.status == 404: raise ControllerNotFoundError(f"{path} not found on compute") return response async def download_image(self, image_type, image): """ Read file of a project and download it :param image_type: Image type :param image: The path of the image :returns: A file stream """ url = self._getUrl(f"/{image_type}/images/{image}") response = await self._session().request("GET", url, auth=self._auth) if response.status == 404: raise ControllerNotFoundError(f"{image} not found on compute") return response async def http_query(self, method, path, data=None, dont_connect=False, stream=False, params=None, **kwargs): """ :param dont_connect: If true do not reconnect if not connected :param stream: If True, return raw aiohttp response for streaming :param params: Optional dict of query parameters to append to the URL """ # a manually disconnected compute is not lazily reconnected either: the # request fails like for any unreachable compute if not self._connected and not self._manual_disconnect and not dont_connect: if self._id == "vm" and not self._controller.gns3vm.running: await self._controller.gns3vm.start() await self.connect() if not self._connected and not dont_connect: raise ComputeError(f"Cannot connect to compute '{self._name}' with request {method} {path}") response = await self._run_http_query(method, path, data=data, stream=stream, params=params, **kwargs) return response async def _try_reconnect(self): """ We catch error during reconnect """ if self._manual_disconnect: return # stop reconnecting, the compute was manually disconnected try: await self.connect() except ControllerError: pass async def _report_connection_failure(self, error): """ Update the connection state after a failure, notify clients and schedule a reconnection attempt with exponential backoff. """ self._connected = False self._last_error = str(error) self._controller.notification.controller_emit("compute.updated", self.asdict()) # Try to reconnect if server unavailable only if not during tests (otherwise we create a ressource usage bomb) if hasattr(sys, "_called_from_test") and sys._called_from_test: return self._connection_failure += 1 # After 10 failures we close the project using the compute to avoid sync issues if self._connection_failure == 10: log.error(f"Could not connect to compute '{self._id}' after multiple attempts: {error}") await self._controller.close_compute_projects(self) # Exponential backoff: 5s, 10s, 20s, 40s, 80s, then cap at 300s delay = min(5 * (2 ** (self._connection_failure - 1)), 300) asyncio.get_event_loop().call_later(delay, lambda: asyncio.ensure_future(self._try_reconnect())) @locking async def connect(self, report_failed_connection=False): """ Check if remote server is accessible """ # an explicit connection request cancels a manual disconnect self._manual_disconnect = False if not self._connected and not self._closed and self.host: try: log.info(f"Connecting to compute '{self._id}'") response = await self._run_http_query("GET", "/capabilities") except ComputeError as e: # Update connection status and notify UI await self._report_connection_failure(e) if report_failed_connection: raise log.warning(f"Cannot connect to compute '{self._id}': {e}") return except (ControllerError, HTTPException) as e: # _run_http_query translates HTTP status errors into ControllerError # subclasses (or a raw HTTPException for unexpected status codes). # They used to escape this method and silently kill the fire-and-forget # connect() task started at controller startup: no notification, no retry. # Schedule the retry, then re-raise so explicit callers still get the error. await self._report_connection_failure(e) log.warning(f"Cannot connect to compute '{self._id}': {e}") raise if "version" not in response.json: msg = f"The server {self._id} is not a GNS3 server" log.error(msg) await self._http_session.close() raise ControllerNotFoundError(msg) self._capabilities = response.json if response.json["version"].split("+")[0] != __version__.split("+")[0]: if self._name.startswith("GNS3 VM"): msg = ( "GNS3 version {} is not the same as the GNS3 VM version {}. Please upgrade the GNS3 VM.".format( __version__, response.json["version"] ) ) else: msg = "GNS3 controller version {} is not the same as compute {} version {}".format( __version__, self._name, response.json["version"] ) if __version_info__[3] == 0: # Stable release log.error(msg) await self._http_session.close() self._last_error = msg raise ControllerError(msg) elif parse_version(__version__)[:2] != parse_version(response.json["version"])[:2]: # We don't allow different major version to interact even with dev build log.error(msg) await self._http_session.close() self._last_error = msg raise ControllerError(msg) else: msg = f"{msg}\nUsing different versions may result in unexpected problems. Please use at your own risk." self._controller.notification.controller_emit("log.warning", {"message": msg}) self._notifications = asyncio.gather(self._connect_notification()) self._connected = True self._connection_failure = 0 self._last_error = None self._controller.notification.controller_emit("compute.updated", self.asdict()) async def disconnect(self): """ Manually disconnect from the compute. The controller stops reconnecting (retry backoff, notification stream reconnect and lazy connection are all blocked) until connect() is called again, or the controller restarts with auto_connect enabled for this compute. """ self._manual_disconnect = True if self._notifications: # Cancelling the notification task runs its finally block: the # compute is marked disconnected and clients are notified. The # flag set above keeps it from scheduling a reconnection. self._notifications.cancel() try: await self._notifications except asyncio.CancelledError: pass self._notifications = None if self._http_session and not self._http_session.closed: await self._http_session.close() if self._connected: self._connected = False self._controller.notification.controller_emit("compute.updated", self.asdict()) async def _connect_notification(self): """ Connect to the notification stream """ ws_url = self._getUrl("/notifications/ws") try: async with self._session().ws_connect(ws_url, auth=self._auth, heartbeat=10) as ws: log.info(f"Connected to compute '{self._id}' WebSocket '{ws_url}'") async for response in ws: if response.type == aiohttp.WSMsgType.TEXT: msg = json.loads(response.data) action = msg.pop("action") event = msg.pop("event") project_id = msg.pop("project_id", None) if action == "ping": self._cpu_usage_percent = event["cpu_usage_percent"] self._memory_usage_percent = event["memory_usage_percent"] self._disk_usage_percent = event["disk_usage_percent"] # FIXME: slow down number of compute events self._controller.notification.controller_emit("compute.updated", self.asdict()) else: await self._controller.notification.dispatch( action, event, project_id=project_id, compute_id=self.id ) else: if response.type == aiohttp.WSMsgType.CLOSE: await ws.close() elif response.type == aiohttp.WSMsgType.ERROR: log.error(f"Error received on compute '{self._id}' WebSocket '{ws_url}': {ws.exception()}") elif response.type == aiohttp.WSMsgType.CLOSED: pass break except asyncio.CancelledError: raise except Exception as e: # A malformed frame or an error raised while dispatching a compute event # used to escape this task (only aiohttp.ClientError was caught) and # permanently killed the notification stream: no more compute.updated # events and no reconnection until the server was restarted. Log the # error with its traceback and reconnect below. log.error(f"Error on compute '{self._id}' notification stream '{ws_url}': {e!r}", exc_info=True) finally: self._connected = False self._cpu_usage_percent = None self._memory_usage_percent = None self._disk_usage_percent = None log.info(f"Connection closed to compute '{self._id}' WebSocket '{ws_url}'") self._controller.notification.controller_emit("compute.updated", self.asdict()) # Try to reconnect after 1 second if server unavailable only if not during tests (otherwise we create a resources usage bomb) from gns3server.api.server import app # The reconnection goes through _try_reconnect (not connect # directly) so that a manual disconnect landing within the 1 # second window is still honoured by the time the timer fires if not self._manual_disconnect and not app.state.exiting and not hasattr(sys, "_called_from_test"): log.info(f"Reconnecting to compute '{self._id}' WebSocket '{ws_url}'") asyncio.get_event_loop().call_later(1, lambda: asyncio.ensure_future(self._try_reconnect())) def _getUrl(self, path): host = self._host # IPV6 if host: # IPV6 if ":" in host: # Reduce IPV6 to his simple form host = str(ipaddress.IPv6Address(host)) if host == "::": host = "::1" host = f"[{host}]" elif host == "0.0.0.0": host = "127.0.0.1" return f"{self._protocol}://{host}:{self._port}/v3/compute{path}" def get_url(self, path): """Returns URL for specific path at Compute""" return self._getUrl(path) async def _run_http_query(self, method, path, data=None, timeout=120, raw=False, stream=False, params=None): async with asynctimeout(delay=timeout): url = self._getUrl(path) headers = {"content-type": "application/json"} chunked = None if data == {}: data = None elif data is not None: if hasattr(data, "asdict"): data = json.dumps(data.asdict()) elif isinstance(data, aiohttp.streams.EmptyStreamReader): data = None # Stream the request elif isinstance(data, aiohttp.streams.StreamReader) or isinstance(data, bytes): chunked = True headers["content-type"] = "application/octet-stream" # Stream from an async iterable (e.g. Starlette request.stream()) elif hasattr(data, "__aiter__"): chunked = True headers["content-type"] = "application/octet-stream" data = await async_iterable_to_stream(data) # If the data is an open file we will iterate on it elif isinstance(data, io.BufferedIOBase): chunked = True headers["content-type"] = "application/octet-stream" else: data = json.dumps(data).encode("utf-8") try: log.debug(f"Attempting request to compute: {method} {url} {headers}") response = await self._session().request( method, url, headers=headers, data=data, auth=self._auth, params=params, chunked=chunked, timeout=timeout, ) except asyncio.TimeoutError: raise ComputeError(f"Timeout error for {method} call to {url} after {timeout}s") except ( aiohttp.ClientError, aiohttp.ServerDisconnectedError, aiohttp.ClientResponseError, ValueError, KeyError, socket.gaierror, ) as e: # aiohttp 2.3.1 raises socket.gaierror when cannot find host raise ComputeError(str(e)) if stream: if response.status >= 300: body = await response.read() msg = body.decode() if body else "" if response.status == 404: raise ControllerNotFoundError(f"{method} {path} not found") elif response.status == 403: raise ControllerForbiddenError(msg) raise ControllerError(f"HTTP {response.status}: {msg}") return response body = await response.read() if body and not raw: body = body.decode() if response.status >= 300: # Try to decode the GNS3 error if body and not raw: try: msg = json.loads(body)["message"] except (KeyError, ValueError): msg = body else: msg = "" if response.status == 400: raise ControllerBadRequestError(msg) elif response.status == 401: raise ControllerUnauthorizedError(f"Invalid authentication for compute '{self.name}' [{self.id}]") elif response.status == 403: raise ControllerForbiddenError(msg) elif response.status == 404: raise ControllerNotFoundError(f"{method} {path} not found") elif response.status == 408 or response.status == 504: raise ControllerTimeoutError(f"{method} {path} request timeout") elif response.status == 409: try: raise ComputeConflictError(url, json.loads(body)) # If the 409 doesn't come from a GNS3 server except ValueError: raise ControllerError(msg) else: raise HTTPException( status_code=response.status, detail=f"HTTP error {response.status} received from compute " f"'{self.name}' for request {method} {path}: {msg}", ) if body and len(body): if raw: response.body = body else: try: response.json = json.loads(body) except ValueError: raise ControllerError(f"The server {self._id} is not a GNS3 server") else: response.json = {} response.body = b"" return response async def get(self, path, **kwargs): return await self.http_query("GET", path, **kwargs) async def post(self, path, data={}, **kwargs): response = await self.http_query("POST", path, data, **kwargs) return response async def put(self, path, data={}, **kwargs): response = await self.http_query("PUT", path, data, **kwargs) return response async def delete(self, path, **kwargs): return await self.http_query("DELETE", path, **kwargs) async def forward(self, method, type, path, data=None): """ Forward a call to the emulator on compute """ try: action = f"/{type}/{path}" res = await self.http_query(method, action, data=data, timeout=None) except aiohttp.ServerDisconnectedError: raise ControllerError(f"Connection lost to {self._id} during {method} {action}") return res.json async def list_files(self, project): """ List files in the project on computes """ path = f"/projects/{project.id}/files" res = await self.http_query("GET", path, timeout=None) return res.json async def get_ip_on_same_subnet(self, other_compute): """ Try to find the best ip for communication from one compute to another :returns: Tuple (ip_for_this_compute, ip_for_other_compute) """ if other_compute == self: return self.host_ip, self.host_ip # Perhaps the user has correct network gateway, we trust him if self.host_ip not in ("0.0.0.0", "127.0.0.1") and other_compute.host_ip not in ("0.0.0.0", "127.0.0.1"): return self.host_ip, other_compute.host_ip this_compute_interfaces = await self.interfaces() other_compute_interfaces = await other_compute.interfaces() # Sort interface to put the compute host in first position # we guess that if user specified this host it could have a reason (VMware Nat / Host only interface) this_compute_interfaces = sorted(this_compute_interfaces, key=lambda i: i["ip_address"] != self.host_ip) other_compute_interfaces = sorted( other_compute_interfaces, key=lambda i: i["ip_address"] != other_compute.host_ip ) for this_interface in this_compute_interfaces: # Skip if no ip or no netmask (vbox when stopped set a null netmask) if len(this_interface["ip_address"]) == 0 or this_interface["netmask"] is None: continue # Ignore 169.254 network because it's for Windows special purpose if this_interface["ip_address"].startswith("169.254."): continue this_network = ipaddress.ip_network( "{}/{}".format(this_interface["ip_address"], this_interface["netmask"]), strict=False ) for other_interface in other_compute_interfaces: if len(other_interface["ip_address"]) == 0 or other_interface["netmask"] is None: continue # Avoid stuff like 127.0.0.1 if other_interface["ip_address"] == this_interface["ip_address"]: continue other_network = ipaddress.ip_network( "{}/{}".format(other_interface["ip_address"], other_interface["netmask"]), strict=False ) if this_network.overlaps(other_network): return this_interface["ip_address"], other_interface["ip_address"] raise ValueError(f"No common subnet for compute {self.name} and {other_compute.name}")