diff --git a/gns3server/handlers/vpcs_handler.py b/gns3server/handlers/vpcs_handler.py
index 78bbbd45..f457b8c7 100644
--- a/gns3server/handlers/vpcs_handler.py
+++ b/gns3server/handlers/vpcs_handler.py
@@ -159,7 +159,8 @@ class VPCSHandler:
vpcs_manager = VPCS.instance()
vm = vpcs_manager.get_vm(request.match_info["uuid"])
- nio = vm.port_add_nio_binding(int(request.match_info["port_id"]), request.json)
+ nio = vpcs_manager.create_nio(vm.vpcs_path, request.json)
+ vm.port_add_nio_binding(int(request.match_info["port_id"]), nio)
response.set_status(201)
response.json(nio)
@@ -191,6 +192,7 @@ class VPCSHandler:
},
status_codes={
204: "VPCS reloaded",
+ 400: "Invalid VPCS instance UUID",
404: "VPCS instance doesn't exist"
},
description="Remove a NIO from a VPCS")
diff --git a/gns3server/modules/attic.py b/gns3server/modules/attic.py
deleted file mode 100644
index 98e2413b..00000000
--- a/gns3server/modules/attic.py
+++ /dev/null
@@ -1,102 +0,0 @@
-# -*- coding: utf-8 -*-
-#
-# Copyright (C) 2014 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 .
-
-"""
-Useful functions... in the attic ;)
-"""
-
-import sys
-import os
-import struct
-import socket
-import stat
-import time
-import aiohttp
-
-import logging
-log = logging.getLogger(__name__)
-
-
-def wait_socket_is_ready(host, port, wait=2.0, socket_timeout=10):
- """
- Waits for a socket to be ready for wait time.
-
- :param host: host/address to connect to
- :param port: port to connect to
- :param wait: maximum wait time
- :param socket_timeout: timeout for the socket
-
- :returns: tuple with boolean indicating if the socket is ready and the last exception
- that occurred when connecting to the socket
- """
-
- # connect to a local address by default
- # if listening to all addresses (IPv4 or IPv6)
- if host == "0.0.0.0":
- host = "127.0.0.1"
- elif host == "::":
- host = "::1"
-
- connection_success = False
- begin = time.time()
- last_exception = None
- while time.time() - begin < wait:
- time.sleep(0.01)
- try:
- with socket.create_connection((host, port), socket_timeout):
- pass
- except OSError as e:
- last_exception = e
- continue
- connection_success = True
- break
-
- return connection_success, last_exception
-
-
-def has_privileged_access(executable):
- """
- Check if an executable can access Ethernet and TAP devices in
- RAW mode.
-
- :param executable: executable path
-
- :returns: True or False
- """
-
- if sys.platform.startswith("win"):
- # do not check anything on Windows
- return True
-
- if os.geteuid() == 0:
- # we are root, so we should have privileged access.
- return True
- if os.stat(executable).st_mode & stat.S_ISUID or os.stat(executable).st_mode & stat.S_ISGID:
- # the executable has set UID bit.
- return True
-
- # test if the executable has the CAP_NET_RAW capability (Linux only)
- if sys.platform.startswith("linux") and "security.capability" in os.listxattr(executable):
- try:
- caps = os.getxattr(executable, "security.capability")
- # test the 2nd byte and check if the 13th bit (CAP_NET_RAW) is set
- if struct.unpack(".
-
+import sys
+import os
+import struct
+import stat
import asyncio
import aiohttp
+import socket
+
+import logging
+log = logging.getLogger(__name__)
from uuid import UUID, uuid4
from ..config import Config
from .project_manager import ProjectManager
+from .nios.nio_udp import NIO_UDP
+from .nios.nio_tap import NIO_TAP
+
class BaseManager:
@@ -147,3 +157,66 @@ class BaseManager:
else:
vm.destroy()
del self._vms[vm.uuid]
+
+ @staticmethod
+ def _has_privileged_access(executable):
+ """
+ Check if an executable can access Ethernet and TAP devices in
+ RAW mode.
+
+ :param executable: executable path
+
+ :returns: True or False
+ """
+
+ if sys.platform.startswith("win"):
+ # do not check anything on Windows
+ return True
+
+ if os.geteuid() == 0:
+ # we are root, so we should have privileged access.
+ return True
+ if os.stat(executable).st_mode & stat.S_ISUID or os.stat(executable).st_mode & stat.S_ISGID:
+ # the executable has set UID bit.
+ return True
+
+ # test if the executable has the CAP_NET_RAW capability (Linux only)
+ if sys.platform.startswith("linux") and "security.capability" in os.listxattr(executable):
+ try:
+ caps = os.getxattr(executable, "security.capability")
+ # test the 2nd byte and check if the 13th bit (CAP_NET_RAW) is set
+ if struct.unpack(".
import pytest
+import aiohttp
import asyncio
import os
from tests.utils import asyncio_patch
@@ -49,7 +50,8 @@ def test_vm_invalid_vpcs_version(loop, project, manager):
with asyncio_patch("gns3server.modules.vpcs.vpcs_vm.VPCSVM._get_vpcs_welcome", return_value="Welcome to Virtual PC Simulator, version 0.1"):
with pytest.raises(VPCSError):
vm = VPCSVM("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", project, manager)
- vm.port_add_nio_binding(0, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
+ nio = manager.create_nio(vm.vpcs_path, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
+ vm.port_add_nio_binding(0, nio)
loop.run_until_complete(asyncio.async(vm.start()))
assert vm.name == "test"
assert vm.uuid == "00010203-0405-0607-0809-0a0b0c0d0e0f"
@@ -59,7 +61,8 @@ def test_vm_invalid_vpcs_version(loop, project, manager):
def test_vm_invalid_vpcs_path(project, manager, loop):
with pytest.raises(VPCSError):
vm = VPCSVM("test", "00010203-0405-0607-0809-0a0b0c0d0e0e", project, manager)
- vm.port_add_nio_binding(0, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
+ nio = manager.create_nio(vm.vpcs_path, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
+ vm.port_add_nio_binding(0, nio)
loop.run_until_complete(asyncio.async(vm.start()))
assert vm.name == "test"
assert vm.uuid == "00010203-0405-0607-0809-0a0b0c0d0e0e"
@@ -68,8 +71,8 @@ def test_vm_invalid_vpcs_path(project, manager, loop):
def test_start(loop, vm):
with asyncio_patch("gns3server.modules.vpcs.vpcs_vm.VPCSVM._check_requirements", return_value=True):
with asyncio_patch("asyncio.create_subprocess_exec", return_value=MagicMock()):
- nio = vm.port_add_nio_binding(0, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
-
+ nio = VPCS.instance().create_nio(vm.vpcs_path, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
+ vm.port_add_nio_binding(0, nio)
loop.run_until_complete(asyncio.async(vm.start()))
assert vm.is_running()
@@ -78,8 +81,8 @@ def test_stop(loop, vm):
process = MagicMock()
with asyncio_patch("gns3server.modules.vpcs.vpcs_vm.VPCSVM._check_requirements", return_value=True):
with asyncio_patch("asyncio.create_subprocess_exec", return_value=process):
- nio = vm.port_add_nio_binding(0, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
-
+ nio = VPCS.instance().create_nio(vm.vpcs_path, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
+ vm.port_add_nio_binding(0, nio)
loop.run_until_complete(asyncio.async(vm.start()))
assert vm.is_running()
loop.run_until_complete(asyncio.async(vm.stop()))
@@ -91,8 +94,8 @@ def test_reload(loop, vm):
process = MagicMock()
with asyncio_patch("gns3server.modules.vpcs.vpcs_vm.VPCSVM._check_requirements", return_value=True):
with asyncio_patch("asyncio.create_subprocess_exec", return_value=process):
- nio = vm.port_add_nio_binding(0, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
-
+ nio = VPCS.instance().create_nio(vm.vpcs_path, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
+ vm.port_add_nio_binding(0, nio)
loop.run_until_complete(asyncio.async(vm.start()))
assert vm.is_running()
loop.run_until_complete(asyncio.async(vm.reload()))
@@ -101,25 +104,29 @@ def test_reload(loop, vm):
def test_add_nio_binding_udp(vm):
- nio = vm.port_add_nio_binding(0, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
+ nio = VPCS.instance().create_nio(vm.vpcs_path, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
+ vm.port_add_nio_binding(0, nio)
assert nio.lport == 4242
def test_add_nio_binding_tap(vm):
- with patch("gns3server.modules.vpcs.vpcs_vm.has_privileged_access", return_value=True):
- nio = vm.port_add_nio_binding(0, {"type": "nio_tap", "tap_device": "test"})
+ with patch("gns3server.modules.base_manager.BaseManager._has_privileged_access", return_value=True):
+ nio = VPCS.instance().create_nio(vm.vpcs_path, {"type": "nio_tap", "tap_device": "test"})
+ vm.port_add_nio_binding(0, nio)
assert nio.tap_device == "test"
def test_add_nio_binding_tap_no_privileged_access(vm):
- with patch("gns3server.modules.vpcs.vpcs_vm.has_privileged_access", return_value=False):
- with pytest.raises(VPCSError):
- vm.port_add_nio_binding(0, {"type": "nio_tap", "tap_device": "test"})
+ with patch("gns3server.modules.base_manager.BaseManager._has_privileged_access", return_value=False):
+ with pytest.raises(aiohttp.web.HTTPForbidden):
+ nio = VPCS.instance().create_nio(vm.vpcs_path, {"type": "nio_tap", "tap_device": "test"})
+ vm.port_add_nio_binding(0, nio)
assert vm._ethernet_adapter.ports[0] is None
def test_port_remove_nio_binding(vm):
- nio = vm.port_add_nio_binding(0, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
+ nio = VPCS.instance().create_nio(vm.vpcs_path, {"type": "nio_udp", "lport": 4242, "rport": 4243, "rhost": "127.0.0.1"})
+ vm.port_add_nio_binding(0, nio)
vm.port_remove_nio_binding(0)
assert vm._ethernet_adapter.ports[0] is None