mirror of
https://github.com/YunoHost/moulinette.git
synced 2024-09-03 20:06:31 +02:00
343 lines
8.5 KiB
Python
343 lines
8.5 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
""" License
|
|
|
|
Copyright (C) 2013 YunoHost
|
|
|
|
This program is free software; you can redistribute it and/or modify
|
|
it under the terms of the GNU Affero 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 Affero General Public License for more details.
|
|
|
|
You should have received a copy of the GNU Affero General Public License
|
|
along with this program; if not, see http://www.gnu.org/licenses
|
|
|
|
"""
|
|
|
|
""" yunohost_firewall.py
|
|
|
|
Manage firewall rules
|
|
"""
|
|
import os
|
|
import sys
|
|
try:
|
|
import miniupnpc
|
|
except ImportError:
|
|
sys.stderr.write('Error: Yunohost CLI Require miniupnpc lib\n')
|
|
sys.exit(1)
|
|
try:
|
|
import yaml
|
|
except ImportError:
|
|
sys.stderr.write('Error: Yunohost CLI Require yaml lib\n')
|
|
sys.stderr.write('apt-get install python-yaml\n')
|
|
sys.exit(1)
|
|
from yunohost import YunoHostError, win_msg
|
|
|
|
|
|
|
|
def firewall_allow(protocol=None, port=None, ipv6=None, upnp=False):
|
|
"""
|
|
Allow connection port/protocol
|
|
|
|
Keyword argument:
|
|
upnp -- upnp
|
|
protocol -- Protocol associated with port
|
|
ipv6 -- ipv6
|
|
port -- Port to open
|
|
|
|
"""
|
|
port = int(port)
|
|
if (upnp):
|
|
add_portmapping(protocol, upnp, ipv6,'a')
|
|
|
|
if 0 < port < 65536:
|
|
if protocol == "Both":
|
|
update_yml(port, 'TCP', 'a', ipv6)
|
|
update_yml(port, 'UDP', 'a', ipv6)
|
|
|
|
else:
|
|
update_yml(port, protocol, 'a', ipv6)
|
|
|
|
win_msg(_("Port successfully openned"))
|
|
|
|
else:
|
|
raise YunoHostError(22, _("Port not between 1 and 65535 : ")+ str(port))
|
|
|
|
return firewall_reload(upnp)
|
|
|
|
|
|
def firewall_disallow(protocol=None, port=None, ipv6=None, upnp=False):
|
|
"""
|
|
Disallow connection
|
|
|
|
Keyword argument:
|
|
upnp -- upnp
|
|
protocol -- Protocol associated with port
|
|
ipv6 -- ipv6
|
|
port -- Port to open
|
|
|
|
"""
|
|
|
|
port = int(port)
|
|
if protocol == "Both":
|
|
update_yml(port, 'TCP', 'r', ipv6)
|
|
update_yml(port, 'UDP', 'r', ipv6)
|
|
else:
|
|
update_yml(port, protocol, 'r', ipv6)
|
|
win_msg(_("Port successfully closed"))
|
|
|
|
return firewall_reload(upnp)
|
|
|
|
|
|
def firewall_list():
|
|
"""
|
|
List all firewall rules
|
|
|
|
|
|
"""
|
|
with open ('firewall.yml') as f:
|
|
firewall = yaml.load(f)
|
|
return firewall
|
|
|
|
def firewall_reload(upnp=False):
|
|
"""
|
|
Reload all firewall rules
|
|
|
|
Keyword argument:
|
|
upnp -- upnp
|
|
|
|
"""
|
|
with open('firewall.yml', 'r') as f:
|
|
firewall = yaml.load(f)
|
|
|
|
os.system ("iptables -P INPUT ACCEPT")
|
|
os.system ("iptables -F")
|
|
os.system ("iptables -X")
|
|
os.system ("iptables -A INPUT -m state --state ESTABLISHED -j ACCEPT")
|
|
|
|
if 22 not in firewall['ipv4']['TCP']:
|
|
update_yml(22, 'TCP', 'a', False)
|
|
|
|
if(os.path.exists("/proc/net/if_inet6")):
|
|
os.system ("ip6tables -P INPUT ACCEPT")
|
|
os.system ("ip6tables -F")
|
|
os.system ("ip6tables -X")
|
|
os.system ("ip6tables -A INPUT -m state --state ESTABLISHED -j ACCEPT")
|
|
|
|
if 22 not in firewall['ipv6']['TCP']:
|
|
update_yml(22, 'TCP', 'a', False)
|
|
|
|
if upnp:
|
|
remove_portmapping()
|
|
|
|
add_portmapping('TCP', upnp, False,'r');
|
|
add_portmapping('UDP', upnp, False,'r');
|
|
|
|
if(os.path.exists("/proc/net/if_inet6")):
|
|
add_portmapping('TCP', upnp, True,'r');
|
|
add_portmapping('UDP', upnp, True,'r');
|
|
|
|
os.system ("iptables -A INPUT -i lo -j ACCEPT")
|
|
os.system ("iptables -A INPUT -p icmp -j ACCEPT")
|
|
os.system ("iptables -P INPUT DROP")
|
|
|
|
if(os.path.exists("/proc/net/if_inet6")):
|
|
os.system ("ip6tables -A INPUT -i lo -j ACCEPT")
|
|
os.system ("ip6tables -A INPUT -p icmp -j ACCEPT")
|
|
os.system ("ip6tables -P INPUT DROP")
|
|
|
|
os.system("service fail2ban restart")
|
|
win_msg(_("Firewall successfully reloaded"))
|
|
|
|
return firewall_list()
|
|
|
|
|
|
def update_yml(port=None, protocol=None, mode=None, ipv6=None):
|
|
"""
|
|
Update firewall.yml
|
|
Keyword arguments:
|
|
protocol -- Protocol used
|
|
port -- Port to open
|
|
mode -- a=append r=remove
|
|
ipv6 -- Boolean ipv6
|
|
|
|
Return
|
|
None
|
|
"""
|
|
if ipv6: ip = 'ipv6'
|
|
else: ip = 'ipv4'
|
|
|
|
with open('firewall.yml', 'r') as f:
|
|
firewall = yaml.load(f)
|
|
|
|
if mode == 'a':
|
|
if port not in firewall[ip][protocol]:
|
|
firewall[ip][protocol].append(port)
|
|
|
|
else:
|
|
raise YunoHostError(22,_("Port already openned :")+ str(port))
|
|
|
|
else:
|
|
if port in firewall[ip][protocol]:
|
|
firewall[ip][protocol].remove(port)
|
|
|
|
else:
|
|
raise YunoHostError(22,_("Port already closed :")+ str(port))
|
|
|
|
firewall[ip][protocol].sort()
|
|
|
|
os.system("mv firewall.yml firewall.yml.old")
|
|
|
|
with open('firewall.yml', 'w') as f:
|
|
yaml.dump(firewall, f)
|
|
|
|
|
|
def add_portmapping(protocol=None, upnp=False, ipv6=None,mode=None,):
|
|
"""
|
|
Send a port mapping rules to igd device
|
|
Keyword arguments:
|
|
protocol -- Protocol used
|
|
upnp -- Boolean upnp
|
|
ipv6 -- Boolean ipv6
|
|
mode -- Add a rule (a) or reload all rules (r)
|
|
|
|
Return
|
|
None
|
|
"""
|
|
if ipv6:
|
|
os.system ("ip6tables -P INPUT ACCEPT")
|
|
else:
|
|
os.system ("iptables -P INPUT ACCEPT")
|
|
|
|
if upnp and mode=='a':
|
|
remove_portmapping()
|
|
|
|
if ipv6: ip = 'ipv6'
|
|
else: ip = 'ipv4'
|
|
with open('firewall.yml', 'r') as f:
|
|
firewall = yaml.load(f)
|
|
|
|
for i,port in enumerate (firewall[ip][protocol]):
|
|
if ipv6:
|
|
os.system ("ip6tables -A INPUT -p "+ protocol +" -i eth0 --dport "+ str(port) +" -j ACCEPT")
|
|
else:
|
|
os.system ("iptables -A INPUT -p "+ protocol +" -i eth0 --dport "+ str(port) +" -j ACCEPT")
|
|
if upnp:
|
|
upnpc = miniupnpc.UPnP()
|
|
upnpc.discoverdelay = 200
|
|
nbigd = upnpc.discover()
|
|
if nbigd:
|
|
upnpc.selectigd()
|
|
upnpc.addportmapping(port, protocol, upnpc.lanaddr, port, 'yunohost firewall : port %u' % port, '')
|
|
|
|
os.system ("iptables -P INPUT DROP")
|
|
|
|
def remove_portmapping():
|
|
"""
|
|
Remove all portmapping rules in the igd device
|
|
Keyword arguments:
|
|
None
|
|
Return
|
|
None
|
|
"""
|
|
upnp = miniupnpc.UPnP()
|
|
upnp.discoverdelay = 200
|
|
nbigd = upnp.discover()
|
|
if nbigd:
|
|
try:
|
|
upnp.selectigd()
|
|
except:
|
|
firewall_reload(False)
|
|
raise YunoHostError(167,_("No upnp devices found"))
|
|
else:
|
|
firewall_reload(False)
|
|
raise YunoHostError(22,_("Can't connect to the igd device"))
|
|
|
|
# list the redirections :
|
|
for i in xrange(100):
|
|
p = upnp.getgenericportmapping(i)
|
|
if p is None: break
|
|
upnp.deleteportmapping(p[0], p[1])
|
|
|
|
|
|
def firewall_installupnp():
|
|
"""
|
|
Add upnp cron
|
|
|
|
|
|
"""
|
|
|
|
with open('firewall.yml', 'r') as f:
|
|
firewall = yaml.load(f)
|
|
|
|
firewall['UPNP']=True;
|
|
|
|
os.system("touch /etc/cron.d/yunohost-firewall")
|
|
os.system("echo '*/50 * * * * root yunohost firewall reload -u>>/dev/null'>/etc/cron.d/yunohost-firewall")
|
|
win_msg(_("UPNP cron installed"))
|
|
|
|
os.system("mv firewall.yml firewall.yml.old")
|
|
|
|
with open('firewall.yml', 'w') as f:
|
|
yaml.dump(firewall, f)
|
|
|
|
|
|
def firewall_removeupnp():
|
|
"""
|
|
Remove upnp cron
|
|
|
|
|
|
"""
|
|
with open('firewall.yml', 'r') as f:
|
|
firewall = yaml.load(f)
|
|
|
|
firewall['UPNP']=False;
|
|
|
|
try:
|
|
os.remove("/etc/cron.d/yunohost-firewall")
|
|
except:
|
|
raise YunoHostError(167,_("UPNP cron was not installed!"))
|
|
|
|
win_msg(_("UPNP cron removed"))
|
|
|
|
os.system("mv firewall.yml firewall.yml.old")
|
|
|
|
with open('firewall.yml', 'w') as f:
|
|
yaml.dump(firewall, f)
|
|
|
|
def firewall_checkupnp():
|
|
"""
|
|
check if UPNP is install or not (0 yes 1 no)
|
|
|
|
|
|
"""
|
|
with open('firewall.yml', 'r') as f:
|
|
firewall = yaml.load(f)
|
|
|
|
if firewall['UPNP']:
|
|
win_msg(_("UPNP is activated"))
|
|
else:
|
|
raise YunoHostError(167,_("UPNP not activated!"))
|
|
|
|
def firewall_stop():
|
|
"""
|
|
Stop iptables and ip6tables
|
|
|
|
|
|
"""
|
|
|
|
os.system ("iptables -P INPUT ACCEPT")
|
|
os.system ("iptables -F")
|
|
os.system ("iptables -X")
|
|
|
|
os.system ("ip6tables -P INPUT ACCEPT")
|
|
os.system ("ip6tables -F")
|
|
os.system ("ip6tables -X")
|
|
if(os.path.exists("/etc/cron.d/yunohost-firewall")):
|
|
firewall_removeupnp()
|
|
|