gns3-server/gns3server/controller/compute.py

#!/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 <http://www.gnu.org/licenses/>.

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}")