OnionSpy/connection.py at main · CoderRony955/OnionSpy · GitHub

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
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
from rich.console import Console
import subprocess
import requests
from pathlib import Path
import sys

from config import (
    tor_control_host,
    tor_control_port,
    tor_data_dir,
    tor_executable,
    tor_socks_host,
    torcc,
    tor_socks_port,
    privoxy_host,
    privoxy_port,
    proxy_url
)

# logger
from logger import logger

console = Console()


class Connection:
    def __init__(self):
        # TOR SOCKS PORT
        self.tor_socks_port = tor_socks_port
        # TOR SOCKS HOST
        self.tor_socks_host = tor_socks_host

        # TOR CONTROL PORT
        self.tor_control_port = tor_control_port
        # TOR CONTROL HOST
        self.tor_control_host = tor_control_host

        self.privoxy_port = privoxy_port  # PRIVOXY PORT
        self.privoxy_host = privoxy_host  # PRIVOXY HOST

        # tor executable
        self.tor_executable = tor_executable

        # tor data dir
        self.tor_data_dir = tor_data_dir

        # torcc (tor config file)
        self.torcc = torcc

        # request proxy url TOR / PRIVOXY
        self.proxy_url = proxy_url


    def verify(self):
        """Check whether user is connected with Tor Network or not
        """
        try:
            # proxies
            proxies = {
                "http": self.proxy_url,
                "https": self.proxy_url
            }

            # make request to check.torproject.org/api/ip to validate tor connection
            response = requests.get(
                url="https://check.torproject.org/api/ip", proxies=proxies, timeout=20)
            # if api response with error code
            if response.status_code != 200:
                console.print(
                    "\n[red]Something went wrong![/red]\n[bold]Please try manually to check wheather tor service running locally or not, or try command \'torverify\' to check again.[/bold]")
                logger.error(
                    f"Tor API response with code {response.status_code} (unable to check connection)")
                return

            # if user is not connected with tor network
            istor = response.json()["IsTor"]
            if not istor:
                logger.error(
                    "Unable to handshake with tor network (not connected with it)")
                return False

            # if user is connected with tor network
            return True

        except KeyboardInterrupt:
            console.print("\n[green]Exit![/green]")
            sys.exit(1)

        except Exception as e:
            logger.error(e)
            return False

    def connect_to_tor(self):
        """Connect user with tor network by starting tor service locally (if user using cloud service then this command does not work)
        """
        try:
            # first of all check connection with tor network
            # if user already connected with tor network (it means tor service running)
            if self.verify():
                return True

            logger.info(
                "Not connected with Tor trying to establish tor connection")
            # if not connected with tor network then connect it
            Path(self.tor_data_dir).mkdir(parents=True, exist_ok=True)

            cmd = [
                self.tor_executable,
                "--SocksPort", str(self.tor_socks_port),
                "--DataDirectory", self.tor_data_dir,
                "-f",
                self.torcc
            ]

            p = subprocess.Popen(
                cmd,
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,
                text=True,
                bufsize=1
            )

            # Wait until Tor is fully bootstrapped
            for line in p.stdout:
                line = line.strip()
                print(line)

                if "Bootstrapped 100%" in line:
                    console.print("[green]Connected to Tor![/green]")
                    logger.info("Connected with Tor Network")
                    return p

        except KeyboardInterrupt:
            console.print("\n[green]Exit![/green]")
            sys.exit(1)

        except Exception as e:
            logger.error(e)
            return False