emdee/stem_examples
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
emdee@spm.plastiras.org 2024-02-04 03:29:52 +00:00
parent b9bdb02067
commit df8ea68e4d
10 changed files with 225 additions and 131 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
Makefile
View File

@ -2,6 +2,7 @@
PREFIX=/usr/local
PYTHON_EXE_MSYS=${PREFIX}/bin/python3.sh
PIP_EXE_MSYS=${PREFIX}/bin/pip3.sh
LOCAL_DOCTEST=${PREFIX}/bin/toxcore_run_doctest3.bash
DOCTEST=${LOCAL_DOCTEST}
MOD=stem_examples
@ -12,6 +13,9 @@ check::
lint::
sh .pylint.sh
xinstall::
${PIP_EXE_MSYS} install --target ${PREFIX}/lib/python3.11/site-packages/ --upgrade .
rsync::
bash .rsync.sh
@ -22,6 +26,8 @@ install::
test::
echo src/${MOD}/check_digests.py
TOR_CONTROLLER_PASSWORD=${PASS} src/${MOD}/check_digests.py
echo src/${MOD}/interpreter.py
TOR_CONTROLLER_PASSWORD=${PASS} src/${MOD}/interpreter.py
echo src/${MOD}/connection_resolution.py
sudo env TOR_CONTROLLER_PASSWORD=${PASS} src/${MOD}/connection_resolution.py
# broken because this site fails: http://128.31.0.39:9131/tor/status-vote

5
setup.cfg
View File

@ -34,9 +34,8 @@ packages=find:
where=src
[options.entry_points]
console_scripts =
phantompy = phantompy.__main__:iMain
exclude_badExits = exclude_badExits:iMain
#console_scripts =
# exclude_badExits = exclude_badExits:iMain
[easy_install]
zip_ok = false

4
src/stem_examples/check_digests.py
View File

@ -110,9 +110,9 @@ if __name__ == '__main__':
else:
sKNOWN_ONION = 'facebookwkhpilnemxj7asaniu7vnjjbiltxjqhye3mhbshg7kx5tfyd' # facebook
LOG.info("Getting some FPs from a sKNOWN_ONION")
from stem_examples.introduction_points import lMain
from stem_examples.introduction_points import lMain as lIPMain
with ignoreStdout():
lArgs = lMain([sKNOWN_ONION])
lArgs = lIPMain([sKNOWN_ONION])
LOG.info(f"Got {len(lArgs)} FPs from a sKNOWN_ONION")
i = iMain(lArgs)

16
src/stem_examples/exit_used.py
View File

@ -20,7 +20,9 @@ from stem_examples.tor_controller import get_controller
global COUNT, IMAX
COUNT=0
IMAX = 0
global IMAX
IMAX = 10
LOG = logging.getLogger()
def stream_event(controller, event):
global COUNT, IMAX
@ -42,7 +44,7 @@ def stream_event(controller, event):
LOG.info("")
def iMain():
def iMain(lArgs=None):
password = os.environ.get('TOR_CONTROLLER_PASSWORD')
if os.path.exists('/run/tor/control'):
@ -50,16 +52,14 @@ def iMain():
else:
controller = get_controller(password=password, port=9051)
if IMAX > 0:
if IMAX <= 0:
LOG.info("Please wait for requests for tor exits. Press 'enter' to end.")
print("")
stream_listener = functools.partial(stream_event, controller)
controller.add_event_listener(stream_listener, EventType.STREAM)
input()
if __name__ == '__main__':
from stem_examples.stem_utils import vsetup_logging
LOG = logging.getLogger()
if len(sys.argv) > 1:
IMAX = int(sys.argv[1])
else:
@ -69,13 +69,17 @@ if __name__ == '__main__':
else:
log_level = 20
vsetup_logging(LOG, log_level)
if len(sys.argv) > 1:
IMAX = int(sys.argv[1])
del sys.argv[1]
try:
iMain()
input()
i = 0
except KeyboardInterrupt as e:
i = 0
except Exception as e:
LOG.exception(f"Exception {e}")
LOG.exception(f"Exception {e}", exc_info=True)
i = 1
sys.exit(i)

57
src/stem_examples/interpreter.py Normal file
View File

@ -0,0 +1,57 @@
#!/usr/local/bin/python3.sh
# -*- mode: python; indent-tabs-mode: nil; py-indent-offset: 4; coding: utf-8 -*-
__doc__ = """List Outdated Relays
Time marches on. Tor makes new releases, and at some point needs to drop
support for old ones. Below is the script we used on ticket 9476 to reach out
to relay operators that needed to upgrade.
https://stem.torproject.org/tutorials/examples/outdated_relays.html
"""
import os
import sys
import logging
import stem.interpreter.commands
from stem.descriptor.remote import DescriptorDownloader
from stem_examples.tor_controller import get_controller
from stem.version import Version
from tor_controller import set_socks_proxy, unset_socks_proxy
LOG = logging.getLogger()
def iMain(lArgs=None):
if not lArgs:
lArgs = ['GETINFO' 'version']
password = os.environ.get('TOR_CONTROLLER_PASSWORD')
if os.path.exists('/run/tor/control'):
controller = get_controller(password=password, unix='/run/tor/control')
else:
controller = get_controller(password=password, port=9051)
interpreter = stem.interpreter.commands.ControlInterpreter(controller)
run_cmd = ' '.join(lArgs)
interpreter.run_command(run_cmd, print_response = True)
return 0
if __name__ == '__main__':
# set_socks_proxy()
from stem_examples.stem_utils import vsetup_logging
if os.environ.get('DEBUG', ''):
log_level = 10
else:
log_level = 20
vsetup_logging(LOG, log_level)
try:
l = iMain(sys.argv[1:])
if l: print(l)
i = 0
except KeyboardInterrupt as e:
i = 0
except Exception as e:
LOG.exception(f"Exception {e}")
i = 1
finally:
unset_socks_proxy()
sys.exit(i)

4
src/stem_examples/mappaddress.py
View File

@ -22,7 +22,6 @@ from stem.control import EventType, Controller
# from tor_controller import set_socks_proxy, unset_socks_proxy
from stem_examples.tor_controller import get_controller
from stem_examples.stem_utils import vsetup_logging
global LOG
import logging
@ -34,7 +33,7 @@ def sMapaddressResolv(target, iPort=9051):
if os.path.exists('/run/tor/control'):
controller = get_controller(password=password, unix='/run/tor/control')
else:
controller = get_controller(password=password, port=9051)
controller = get_controller(password=password, port=iPort)
map_dict = {"0.0.0.0": target}
map_ret = controller.map_address(map_dict)
@ -44,6 +43,7 @@ def sMapaddressResolv(target, iPort=9051):
LOG.exception(e)
if __name__ == '__main__':
from stem_examples.stem_utils import vsetup_logging
if os.environ.get('DEBUG', ''):
log_level = 10
else:

4
src/stem_examples/outdated_relays.py
View File

@ -20,7 +20,7 @@ from tor_controller import set_socks_proxy, unset_socks_proxy
LOG = logging.getLogger()
def iMain():
def iMain(lArgs=None):
downloader = DescriptorDownloader(use_mirrors=True)
count, with_contact = 0, 0
elts = downloader.get_server_descriptors()
@ -47,7 +47,7 @@ if __name__ == '__main__':
log_level = 20
vsetup_logging(LOG, log_level)
try:
l = iMain()
l = iMain([])
if l: print(l)
i = 0
except KeyboardInterrupt as e:

9
src/stem_examples/relay_connections.py
View File

@ -21,14 +21,14 @@ import stem.connection
import stem.util.system
import stem.util.str_tools
from stem.control import Listener
from stem.control import Controller
from stem.control import Listener, Controller
from stem.util.connection import get_connections, port_usage, is_valid_ipv4_address
from stem_examples.tor_controller import get_controller
global LOG
import logging
global LOG
LOG = logging.getLogger()
HEADER_LINE = " {version} uptime: {uptime} flags: {flags}\n"
@ -154,7 +154,6 @@ def iMain(lArgs=None):
print(DIV)
print('')
if __name__ == '__main__':
from stem_examples.stem_utils import vsetup_logging
LOG = logging.getLogger()
@ -165,7 +164,7 @@ if __name__ == '__main__':
vsetup_logging(LOG, log_level)
LOG.setLevel(logging.DEBUG)
try:
l = iMain()
l = iMain([])
if l: print(l)
i = 0
except KeyboardInterrupt as e:

3
src/stem_examples/tor_bootstrap_check.py
View File

@ -60,9 +60,8 @@ if __name__ == '__main__':
else:
log_level = 20
vsetup_logging(LOG, log_level)
LOG.setLevel(logging.DEBUG)
try:
i = iMain()
i = iMain([])
except KeyboardInterrupt as e:
i = 0
except Exception as e:

76
stem_examples.txt
View File

@ -55,19 +55,20 @@ If you don't have one, make it with the settings from your torrc:
>>> import yaml
>>> try:
... sFacts = open('/usr/local/etc/testforge/testforge.yml').read()
... assert sFacts, sFacts
... except:
... dFacts = {
... HTTPS_PROXYHOST: "127.0.0.1",
... HTTPS_PROXYPORT: 9128,
... HTTPS_PROXYTYPE: "http",
... SOCKS_PROXYHOST: "127.0.0.1",
... SOCKS_PROXYPORT: 9050,
... SOCKS_PROXYTYPE: "socks5",
... }
... dFacts = dict(
... HTTPS_PROXYHOST="127.0.0.1",
... HTTPS_PROXYPORT=9128,
... HTTPS_PROXYTYPE="http",
... SOCKS_PROXYHOST="127.0.0.1",
... SOCKS_PROXYPORT=9050,
... SOCKS_PROXYTYPE="socks5",
... )
... else:
... assert sFacts
... dFacts = yaml.safe_load(sFacts)
FixMe: use the settings for the ports and directories below.
>>> import os
@ -94,7 +95,7 @@ and then the descriptor-signing-key-cert:
>>> introduction_points.iMain(lKNOWN_ONIONS) #doctest: +ELLIPSIS +NORMALIZE_WHITESPACE
0
hs-descriptor 3
hs-descriptor 3
descriptor-lifetime ...
<BLANKLINE>
@ -106,11 +107,17 @@ How can you figure out what exit you're using?
>>> print("exit_used", file=sys.stderr)
>>> import exit_used
>>> exit_used.iMain([])
## relay_connections Connection Summary
>>> print("relay_connections", file=sys.stderr)
>>> import relay_connections
>>> relay_connections.iMain([]) #doctest: +ELLIPSIS +NORMALIZE_WHITESPACE
+------------------------------+------+------+
...
+------------------------------+------+------+
<BLANKLINE>
The following provides a summary of your relay's inbound and outbound connections.
You must be root or tor to run this:
@ -124,25 +131,22 @@ provides an easy method for accessing this information.
>>> print("connection_resolution", file=sys.stderr)
>>> import connection_resolution
>>> connection_resolution.iMain() #doctest: +ELLIPSIS +NORMALIZE_WHITESPACE
>>> connection_resolution.iMain([]) #doctest: +ELLIPSIS +NORMALIZE_WHITESPACE
0
INFO Our platform supports connection resolution via: ...
<BLANKLINE>
## outdated_relays List Outdated Relays
## interpreter
Time marches on. Tor makes new releases, and at some point needs to drop
support for old ones. Below is the script we used on ticket 9476 to reach out
to relay operators that needed to upgrade.
>>> print("outdated_relays", file=sys.stderr)
>>> import outdated_relays
>>> outdated_relays.iMain() #doctest: +ELLIPSIS +NORMALIZE_WHITESPACE
0
Checking for outdated relays ...
>>> print("interpreter", file=sys.stderr)
>>> import interpreter
>>> lArgs = ['GETINFO', 'version']
>>> interpreter.iMain(lArgs) #doctest: +ELLIPSIS +NORMALIZE_WHITESPACE
250-version=0.4.8.10
250 OK
<BLANKLINE>
0
## tor_bootstrap_check
@ -152,9 +156,35 @@ to relay operators that needed to upgrade.
A script by adrelanos@riseup.net to check what percentage of boostrapping
tor is at. This fails under doctest but not from the cmdline
>>> tor_bootstrap_check.iMain() #doctest: +ELLIPSIS +NORMALIZE_WHITESPACE
>>> tor_bootstrap_check.iMain([]) #doctest: +ELLIPSIS +NORMALIZE_WHITESPACE
0
NOTICE ...
<BLANKLINE>
## check_digests
>>> print("check_digests", file=sys.stderr)
>>> from support_testing import ignoreStdout
>>> from check_digests import iMain
>>> sKNOWN_ONION = 'facebookwkhpilnemxj7asaniu7vnjjbiltxjqhye3mhbshg7kx5tfyd' # facebook
>>> from stem_examples.introduction_points import lMain as lIPMain
>>> with ignoreStdout():
... lArgs = lIPMain([sKNOWN_ONION])
>>> iMain(lArgs)
0
## outdated_relays List Outdated Relays
Time marches on. Tor makes new releases, and at some point needs to drop
support for old ones. Below is the script we used on ticket 9476 to reach out
to relay operators that needed to upgrade.
>>> print("outdated_relays", file=sys.stderr)
>>> import outdated_relays
>>> outdated_relays.iMain([]) #doctest: +ELLIPSIS +NORMALIZE_WHITESPACE
0
Checking for outdated relays ...
<BLANKLINE>