summaryrefslogtreecommitdiff
path: root/src/leap/bitmask/vpn/udstelnet.py
blob: e6c82350d64fd5486368467536cba4da15ec87f4 (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
# -*- coding: utf-8 -*-
# udstelnet.py
# Copyright (C) 2013 LEAP
#
# 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/>.

import os
import socket
import telnetlib


class ConnectionRefusedError(Exception):
    pass


class MissingSocketError(Exception):
    pass


class UDSTelnet(telnetlib.Telnet):
    """
    A telnet-alike class, that can listen on unix domain sockets
    """

    def open(self, host, port=23, timeout=socket._GLOBAL_DEFAULT_TIMEOUT):
        """
        Connect to a host. If port is 'unix', it will open a
        connection over unix docmain sockets.

        The optional second argument is the port number, which
        defaults to the standard telnet port (23).
        Don't try to reopen an already connected instance.
        """
        self.eof = 0
        self.host = host
        self.port = port
        self.timeout = timeout

        if self.port == "unix":
            # unix sockets spoken
            if not os.path.exists(self.host):
                raise MissingSocketError()
            self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
            try:
                self.sock.connect(self.host)
            except socket.error:
                raise ConnectionRefusedError()
        else:
            self.sock = socket.create_connection((host, port), timeout)