gaschz/qubes-network-server
1
0
Fork 0
mirror of https://github.com/Rudd-O/qubes-network-server.git synced 2025-03-01 14:22:35 +01:00
Code Issues Packages Projects Releases Wiki Activity

Tag 0.0.18.

Browse Source
This commit is contained in:
Manuel Amador (Rudd-O) 2021-10-29 00:43:28 +00:00
parent 21c09202cc
commit 581a913999
2 changed files with 22 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
qubes-network-server.spec
View File

@ -3,7 +3,7 @@
%define mybuildnumber %{?build_number}%{?!build_number:1} %define mybuildnumber %{?build_number}%{?!build_number:1}
Name: qubes-network-server Name: qubes-network-server
Version: 0.0.17 Version: 0.0.18
Release: %{mybuildnumber}%{?dist} Release: %{mybuildnumber}%{?dist}
Summary: Turn your Qubes OS into a network server Summary: Turn your Qubes OS into a network server
BuildArch: noarch BuildArch: noarch

45
qubesnetworkserver/__init__.py
View File

@ -2,23 +2,25 @@ import qubes.ext
def l(text, *parms): def l(text, *parms):
return # This exists only to debug.
if parms: if parms:
text = text % parms text = text % parms
import sys import sys
print("nsext:", text, file=sys.stderr) print("nsext:", text, file=sys.stderr)
sys.stderr.flush() sys.stderr.flush()
l("loaded") l("loaded")
class QubesNetworkServerExtension(qubes.ext.Extension):
class QubesNetworkServerExtension(qubes.ext.Extension):
def shutdown_routing_for_vm(self, netvm, appvm): def shutdown_routing_for_vm(self, netvm, appvm):
l("shutdown routing for vm %s %s", netvm, appvm) l("shutdown routing for vm %s %s", netvm, appvm)
self.reload_routing_for_vm(netvm, appvm, True) self.reload_routing_for_vm(netvm, appvm, True)
def reload_routing_for_vm(self, netvm, appvm, shutdown=False): def reload_routing_for_vm(self, netvm, appvm, shutdown=False):
'''Reload the routing method for the VM.''' """Reload the routing method for the VM."""
l("reload routing for vm %s %s shutdown %s", netvm, appvm, shutdown) l("reload routing for vm %s %s shutdown %s", netvm, appvm, shutdown)
if not netvm.is_running(): if not netvm.is_running():
return return
@ -30,7 +32,7 @@ class QubesNetworkServerExtension(qubes.ext.Extension):
self.setup_forwarding_for_vm(netvm, appvm, ip, remove=shutdown) self.setup_forwarding_for_vm(netvm, appvm, ip, remove=shutdown)
def setup_forwarding_for_vm(self, netvm, appvm, ip, remove=False): def setup_forwarding_for_vm(self, netvm, appvm, ip, remove=False):
''' """
Record in Qubes DB that the passed VM may be meant to have traffic Record in Qubes DB that the passed VM may be meant to have traffic
forwarded to and from it, rather than masqueraded from it and blocked forwarded to and from it, rather than masqueraded from it and blocked
to it. to it.
@ -46,46 +48,41 @@ class QubesNetworkServerExtension(qubes.ext.Extension):
If `remove` is True, then we remove the respective routing method from If `remove` is True, then we remove the respective routing method from
the Qubes DB instead. the Qubes DB instead.
''' """
l("setup forwarding for vm vm %s %s %s remove %s", netvm, appvm, ip, remove) l("setup forwarding for vm vm %s %s %s remove %s", netvm, appvm, ip, remove)
if ip is None: if ip is None:
return return
routing_method = appvm.features.check_with_template( routing_method = appvm.features.check_with_template(
'routing-method', 'masquerade' "routing-method", "masquerade"
) )
base_file = '/qubes-routing-method/{}'.format(ip) base_file = "/qubes-routing-method/{}".format(ip)
if remove: if remove:
netvm.untrusted_qdb.rm(base_file) netvm.untrusted_qdb.rm(base_file)
elif routing_method == 'forward': elif routing_method == "forward":
netvm.untrusted_qdb.write(base_file, 'forward') netvm.untrusted_qdb.write(base_file, "forward")
else: else:
netvm.untrusted_qdb.write(base_file, 'masquerade') netvm.untrusted_qdb.write(base_file, "masquerade")
@qubes.ext.handler( @qubes.ext.handler(
'domain-feature-set:routing-method', "domain-feature-set:routing-method",
'domain-feature-delete:routing-method', "domain-feature-delete:routing-method",
) )
def on_routing_method_changed( def on_routing_method_changed(self, vm, ignored_feature, **kwargs):
self,
vm,
ignored_feature,
**kwargs
):
# pylint: disable=no-self-use,unused-argument # pylint: disable=no-self-use,unused-argument
l("routing method changed %s", vm) l("routing method changed %s", vm)
if 'oldvalue' not in kwargs or kwargs.get('oldvalue') != kwargs.get('value'): if "oldvalue" not in kwargs or kwargs.get("oldvalue") != kwargs.get("value"):
if vm.netvm: if vm.netvm:
self.reload_routing_for_vm(vm.netvm, vm) self.reload_routing_for_vm(vm.netvm, vm)
@qubes.ext.handler('domain-qdb-create') @qubes.ext.handler("domain-qdb-create")
def on_domain_qdb_create(self, vm, event, **kwargs): def on_domain_qdb_create(self, vm, event, **kwargs):
''' Fills the QubesDB with firewall entries. ''' """ Fills the QubesDB with firewall entries. """
# pylint: disable=unused-argument # pylint: disable=unused-argument
l("domain create %s %s", vm, event) l("domain create %s %s", vm, event)
if vm.netvm: if vm.netvm:
self.reload_routing_for_vm(vm.netvm, vm) self.reload_routing_for_vm(vm.netvm, vm)
@qubes.ext.handler('domain-start') @qubes.ext.handler("domain-start")
def on_domain_started(self, vm, event, **kwargs): def on_domain_started(self, vm, event, **kwargs):
# pylint: disable=unused-argument # pylint: disable=unused-argument
l("domain started %s %s", vm, event) l("domain started %s %s", vm, event)
@ -95,7 +92,7 @@ class QubesNetworkServerExtension(qubes.ext.Extension):
except AttributeError: except AttributeError:
pass pass
@qubes.ext.handler('domain-shutdown') @qubes.ext.handler("domain-shutdown")
def on_domain_shutdown(self, vm, event, **kwargs): def on_domain_shutdown(self, vm, event, **kwargs):
# pylint: disable=unused-argument # pylint: disable=unused-argument
l("domain shutdown %s %s", vm, event) l("domain shutdown %s %s", vm, event)
@ -106,8 +103,8 @@ class QubesNetworkServerExtension(qubes.ext.Extension):
pass pass
if vm.netvm: if vm.netvm:
self.shutdown_routing_for_vm(vm.netvm, vm) self.shutdown_routing_for_vm(vm.netvm, vm)
@qubes.ext.handler('net-domain-connect') @qubes.ext.handler("net-domain-connect")
def on_net_domain_connect(self, vm, event): def on_net_domain_connect(self, vm, event):
# pylint: disable=unused-argument # pylint: disable=unused-argument
l("domain connect %s %s", vm, event) l("domain connect %s %s", vm, event)