summaryrefslogtreecommitdiff
path: root/kvmd/apps/otgnet/netctl.py
blob: 874e904aadfb1722031171d3a33d87507eb2d7ef (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
# ========================================================================== #
#                                                                            #
#    KVMD - The main Pi-KVM daemon.                                          #
#                                                                            #
#    Copyright (C) 2018  Maxim Devaev <[email protected]>                    #
#                                                                            #
#    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 <https://www.gnu.org/licenses/>.  #
#                                                                            #
# ========================================================================== #


from typing import List
from typing import Dict


# =====
class BaseCtl:
    def get_command(self, direct: bool) -> List[str]:
        raise NotImplementedError


class IfaceUpCtl(BaseCtl):
    def __init__(self, base_cmd: List[str], iface: str) -> None:
        self.__base_cmd = base_cmd
        self.__iface = iface

    def get_command(self, direct: bool) -> List[str]:
        return [*self.__base_cmd, "link", "set", self.__iface, ("up" if direct else "down")]


class IfaceAddIpCtl(BaseCtl):
    def __init__(self, base_cmd: List[str], iface: str, cidr: str) -> None:
        self.__base_cmd = base_cmd
        self.__iface = iface
        self.__cidr = cidr

    def get_command(self, direct: bool) -> List[str]:
        return [*self.__base_cmd, "address", ("add" if direct else "del"), self.__cidr, "dev", self.__iface]


class IptablesDropAllCtl(BaseCtl):
    def __init__(self, base_cmd: List[str], iface: str) -> None:
        self.__base_cmd = base_cmd
        self.__iface = iface

    def get_command(self, direct: bool) -> List[str]:
        return [*self.__base_cmd, ("-A" if direct else "-D"), "INPUT", "-i", self.__iface, "-j", "DROP"]


class IptablesAllowPortCtl(BaseCtl):
    def __init__(self, base_cmd: List[str], iface: str, port: int, tcp: bool) -> None:
        self.__base_cmd = base_cmd
        self.__iface = iface
        self.__port = port
        self.__proto = ("tcp" if tcp else "udp")

    def get_command(self, direct: bool) -> List[str]:
        return [
            *self.__base_cmd,
            ("-A" if direct else "-D"), "INPUT", "-i", self.__iface, "-p", self.__proto,
            "--dport", str(self.__port), "-j", "ACCEPT",
        ]


class CustomCtl(BaseCtl):
    def __init__(
        self,
        direct_cmd: List[str],
        reverse_cmd: List[str],
        placeholders: Dict[str, str],
    ) -> None:

        self.__direct_cmd = direct_cmd
        self.__reverse_cmd = reverse_cmd
        self.__placeholders = placeholders

    def get_command(self, direct: bool) -> List[str]:
        return [
            part.format(**self.__placeholders)
            for part in (self.__direct_cmd if direct else self.__reverse_cmd)
        ]