arno/gns3-server
1
0
Fork 0
mirror of https://github.com/GNS3/gns3-server synced 2024-11-24 09:18:08 +00:00
Code Issues Releases Wiki Activity

Moves NIO creation to the base manager.

Browse Source
This commit is contained in:
Jeremy 2015-01-22 18:04:24 -07:00
parent 6ec4dea9b9
commit 2681defe27
8 changed files with 126 additions and 156 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
gns3server/handlers/vpcs_handler.py
View File

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

102
gns3server/modules/attic.py
View File

@ -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 <http://www.gnu.org/licenses/>.
"""
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("<IIIII", caps)[1] & 1 << 13:
return True
except Exception as e:
log.error("could not determine if CAP_NET_RAW capability is set for {}: {}".format(executable, e))
return False

75
gns3server/modules/base_manager.py
View File

@ -15,14 +15,24 @@
# 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 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("<IIIII", caps)[1] & 1 << 13:
return True
except Exception as e:
log.error("could not determine if CAP_NET_RAW capability is set for {}: {}".format(executable, e))
return False
def create_nio(self, executable, nio_settings):
"""
Creates a new NIO.
:param nio_settings: information to create the NIO
:returns: a NIO object
"""
nio = None
if nio_settings["type"] == "nio_udp":
lport = nio_settings["lport"]
rhost = nio_settings["rhost"]
rport = nio_settings["rport"]
try:
# TODO: handle IPv6
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
sock.connect((rhost, rport))
except OSError as e:
raise aiohttp.web.HTTPInternalServerError(text="Could not create an UDP connection to {}:{}: {}".format(rhost, rport, e))
nio = NIO_UDP(lport, rhost, rport)
elif nio_settings["type"] == "nio_tap":
tap_device = nio_settings["tap_device"]
if not self._has_privileged_access(executable):
raise aiohttp.web.HTTPForbidden(text="{} has no privileged access to {}.".format(executable, tap_device))
nio = NIO_TAP(tap_device)
assert nio is not None
return nio

3
gns3server/modules/nios/nio_tap.py
View File

@ -48,4 +48,5 @@ class NIO_TAP(object):
def __json__(self):
return {"type": "nio_tap", "tap_device": self._tap_device}
return {"type": "nio_tap",
"tap_device": self._tap_device}

5
gns3server/modules/nios/nio_udp.py
View File

@ -74,4 +74,7 @@ class NIO_UDP(object):
def __json__(self):
return {"type": "nio_udp", "lport": self._lport, "rport": self._rport, "rhost": self._rhost}
return {"type": "nio_udp",
"lport": self._lport,
"rport": self._rport,
"rhost": self._rhost}

42
gns3server/modules/vpcs/vpcs_vm.py
View File

@ -26,18 +26,14 @@ import subprocess
import signal
import re
import asyncio
import socket
import shutil
from pkg_resources import parse_version
from .vpcs_error import VPCSError
from ..adapters.ethernet_adapter import EthernetAdapter
from ..nios.nio_udp import NIO_UDP
from ..nios.nio_tap import NIO_TAP
from ..attic import has_privileged_access
from ..base_vm import BaseVM
import logging
log = logging.getLogger(__name__)
@ -116,6 +112,16 @@ class VPCSVM(BaseVM):
"script_file": self.script_file,
"startup_script": self.startup_script}
@property
def vpcs_path(self):
"""
Returns the VPCS executable path.
:returns: path to VPCS
"""
return self._path
@property
def console(self):
"""
@ -303,7 +309,7 @@ class VPCSVM(BaseVM):
return True
return False
def port_add_nio_binding(self, port_id, nio_settings):
def port_add_nio_binding(self, port_id, nio):
"""
Adds a port NIO binding.
@ -315,26 +321,6 @@ class VPCSVM(BaseVM):
raise VPCSError("Port {port_id} doesn't exist in adapter {adapter}".format(adapter=self._ethernet_adapter,
port_id=port_id))
nio = None
if nio_settings["type"] == "nio_udp":
lport = nio_settings["lport"]
rhost = nio_settings["rhost"]
rport = nio_settings["rport"]
try:
# TODO: handle IPv6
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
sock.connect((rhost, rport))
except OSError as e:
raise VPCSError("Could not create an UDP connection to {}:{}: {}".format(rhost, rport, e))
nio = NIO_UDP(lport, rhost, rport)
elif nio_settings["type"] == "nio_tap":
tap_vm = nio_settings["tap_device"]
if not has_privileged_access(self._path):
raise VPCSError("{} has no privileged access to {}.".format(self._path, tap_vm))
nio = NIO_TAP(tap_vm)
if not nio:
raise VPCSError("Requested NIO does not exist or is not supported: {}".format(nio_settings["type"]))
self._ethernet_adapter.add_nio(port_id, nio)
log.info("VPCS {name} {uuid}]: {nio} added to port {port_id}".format(name=self._name,
uuid=self.uuid,
@ -404,13 +390,13 @@ class VPCSVM(BaseVM):
nio = self._ethernet_adapter.get_nio(0)
if nio:
if isinstance(nio, NIO_UDP):
if str(nio) == "NIO UDP":
# UDP tunnel
command.extend(["-s", str(nio.lport)]) # source UDP port
command.extend(["-c", str(nio.rport)]) # destination UDP port
command.extend(["-t", nio.rhost]) # destination host
elif isinstance(nio, NIO_TAP):
elif str(nio) == "NIO TAP":
# TAP interface
command.extend(["-e"])
command.extend(["-d", nio.tap_vm])

4
tests/api/test_vpcs.py
View File

@ -86,8 +86,8 @@ def test_vpcs_nio_create_udp(server, vm):
assert response.json["type"] == "nio_udp"
@patch("gns3server.modules.vpcs.vpcs_vm.has_privileged_access", return_value=True)
def test_vpcs_nio_create_tap(mock, server, vm):
def test_vpcs_nio_create_tap(server, vm):
with patch("gns3server.modules.base_manager.BaseManager._has_privileged_access", return_value=True):
response = server.post("/vpcs/{}/ports/0/nio".format(vm["uuid"]), {"type": "nio_tap",
"tap_device": "test"})
assert response.status == 201

37
tests/modules/vpcs/test_vpcs_vm.py
View File

@ -16,6 +16,7 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
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