(
self,
mode: MODE,
port: int,
cls: Type[Packet],
af: socket.AddressFamily = socket.AF_INET,
proto: socket.SocketKind = socket.SOCK_STREAM,
remote_address: str = None,
remote_af: Optional[socket.AddressFamily] = None,
bind_address: str = None,
tls: bool = False,
crtfile: Optional[str] = None,
keyfile: Optional[str] = None,
timeout: int = 2,
MTU: int = MTU,
**kwargs,
)
| 99 | TPROXY = 1 |
| 100 | |
| 101 | def __init__( |
| 102 | self, |
| 103 | mode: MODE, |
| 104 | port: int, |
| 105 | cls: Type[Packet], |
| 106 | af: socket.AddressFamily = socket.AF_INET, |
| 107 | proto: socket.SocketKind = socket.SOCK_STREAM, |
| 108 | remote_address: str = None, |
| 109 | remote_af: Optional[socket.AddressFamily] = None, |
| 110 | bind_address: str = None, |
| 111 | tls: bool = False, |
| 112 | crtfile: Optional[str] = None, |
| 113 | keyfile: Optional[str] = None, |
| 114 | timeout: int = 2, |
| 115 | MTU: int = MTU, |
| 116 | **kwargs, |
| 117 | ): |
| 118 | self.mode = mode |
| 119 | self.port = port |
| 120 | self.cls = cls |
| 121 | self.af = af |
| 122 | self.remote_af = remote_af if remote_af is not None else af |
| 123 | self.proto = proto |
| 124 | self.tls = tls |
| 125 | self.crtfile = crtfile |
| 126 | self.keyfile = keyfile |
| 127 | self.timeout = timeout |
| 128 | self.MTU = MTU |
| 129 | self.remote_address = remote_address |
| 130 | if self.tls or self.af == 40: # TLS or VSOCK |
| 131 | self.sockcls = StreamSocketPeekless |
| 132 | else: |
| 133 | self.sockcls = StreamSocket |
| 134 | # Chose 'bind_address' depending on the mode |
| 135 | self.bind_address = bind_address |
| 136 | if self.bind_address is None: |
| 137 | if self.mode == ForwardMachine.MODE.SERVER: |
| 138 | self.bind_address = "0.0.0.0" |
| 139 | elif self.mode == ForwardMachine.MODE.TPROXY: |
| 140 | self.bind_address = "2.2.2.2" |
| 141 | else: |
| 142 | raise ValueError("Unknown mode :/") |
| 143 | red = lambda z: functools.reduce(lambda x, y: x + y, z) |
| 144 | # Utils |
| 145 | self.ct = DefaultTheme() |
| 146 | self.local_ips = red(red(list(x.ips.values())) for x in conf.ifaces.values()) |
| 147 | self.cache = {} |
| 148 | super(ForwardMachine, self).__init__(**kwargs) |
| 149 | |
| 150 | def run(self): |
| 151 | """ |
nothing calls this directly
no test coverage detected