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

update

This commit is contained in:
emdee@spm.plastiras.org 2024-01-17 16:30:48 +00:00
parent 72010d8f8d
commit 91465abf2a
12 changed files with 120 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
Makefile
View File

@ -1,4 +1,6 @@
gggggPREFIX=/usr/local
# to run the tests, run make PASS=controllerpassword test
PREFIX=/usr/local
PYTHON_EXE_MSYS=${PREFIX}/bin/python3.sh
LOCAL_DOCTEST=${PREFIX}/bin/toxcore_run_doctest3.bash
DOCTEST=${LOCAL_DOCTEST}
@ -13,22 +15,29 @@ lint::
rsync::
bash .rsync.sh
pyi::
echo FixMe
install::
${PYTHON_EXE_MSYS} -m pip --timeout=30 --disable-pip-version-check --proxy http://127.0.0.1:9128 install --only-binary :none: --progress-bar=off --target /usr/local/lib/python3.11/site-packages --upgrade .
# execute these tests as: make test PASS=password
test::
export TOR_CONTROLLER_PASSWORD=${PASS}
${PYTHON_EXE_MSYS} src/${MOD}/check_digests.py
echo src/${MOD}/check_digests.py
TOR_CONTROLLER_PASSWORD=${PASS} src/${MOD}/check_digests.py
# broken because this site fails: http://128.31.0.39:9131/tor/status-vote
# ${PYTHON_EXE_MSYS} src/${MOD}/compare_flags.py
# cant use from make: waits for the cmdline to to terminate
# TOR_CONTROLLER_PASSWORD=${PASS} ${PYTHON_EXE_MSYS} src/${MOD}/exit_used.py
# TOR_CONTROLLER_PASSWORD=${PASS} ${PYTHON_EXE_MSYS} src/${MOD}/exit_used.py 10
echo src/${MOD}/introduction_points.py
TOR_CONTROLLER_PASSWORD=${PASS} ${PYTHON_EXE_MSYS} src/${MOD}/introduction_points.py
TOR_CONTROLLER_PASSWORD=${PASS} ${PYTHON_EXE_MSYS} src/${MOD}/list_circuits.py
echo src/${MOD}/list_circuits.py
TOR_CONTROLLER_PASSWORD=${PASS} ${PYTHON_EXE_MSYS} src/${MOD}/list_circuits.py
echo src/${MOD}/mappaddress.py
TOR_CONTROLLER_PASSWORD=${PASS} ${PYTHON_EXE_MSYS} src/${MOD}/mappaddress.py
#hangs TOR_CONTROLLER_PASSWORD=${PASS} ${PYTHON_EXE_MSYS} src/${MOD}/outdated_relays.py
echo src/${MOD}/outdated_relays.py NOT WORKING?
TOR_CONTROLLER_PASSWORD=${PASS} ${PYTHON_EXE_MSYS} src/${MOD}/outdated_relays.py
echo src/${MOD}/relay_connections.py
TOR_CONTROLLER_PASSWORD=${PASS} ${PYTHON_EXE_MSYS} src/${MOD}/relay_connections.py
#wrosk TOR_CONTROLLER_PASSWORD=${PASS} ${PYTHON_EXE_MSYS} src/${MOD}/tor_bootstrap_check.py
echo src/${MOD}/tor_bootstrap_check.py
TOR_CONTROLLER_PASSWORD=${PASS} ${PYTHON_EXE_MSYS} src/${MOD}/tor_bootstrap_check.py
doctest:
export PYTHONPATH=${PWD}

6
README.md
View File

@ -57,6 +57,12 @@ https://stem.torproject.org/tutorials/examples/list_circuits.html
### mappaddress
Mappaddress queries the socks proxy with an onion address and returns the
IP address that it will use for it. the address will be in the block specified
by the VirtualAddrNetworkIPv4 setting of the torrc, e.g.
VirtualAddrNetworkIPv4 172.16.0.0/12
### outdated_relays List Outdated Relays
Time marches on. Tor makes new releases, and at some point needs to

1
src/stem_examples/__init__.py
View File

@ -1 +1,2 @@
__version__ = "1.0.0"

20
src/stem_examples/check_digests.py
View File

@ -1,3 +1,4 @@
#!/usr/local/bin/python3.sh
# -*-mode: python; py-indent-offset: 2; indent-tabs-mode: nil; coding: utf-8-unix -*-
__doc__ = """
@ -8,15 +9,16 @@ Tor relay information is provided by multiple documents. Signed descriptors tran
Stem can calculate digests from server, extrainfo, microdescriptor, and
consensus documents. For instance, to validate an extrainfo descriptor...
https://stem.torproject.org/tutorials/examples/check_digests.html """
https://stem.torproject.org/tutorials/examples/check_digests.html
"""
import os
import sys
import contextlib
import logging
import stem
from stem_examples.tor_controller import set_socks_proxy
from stem_examples.stem_utils import vsetup_logging
LOG = logging.getLogger()
@ -93,23 +95,15 @@ def iMain(lArgs=None):
return iRetval
if __name__ == '__main__':
from stem_examples.stem_utils import vsetup_logging
if os.environ.get('DEBUG', ''):
log_level = 10
logging.getLogger('stem').setLevel(20)
else:
log_level = 20
logging.getLogger('stem').setLevel(30)
vsetup_logging(LOG, log_level)
try:
logging.getLogger('stem').setLevel(20)
import stem.descriptor.remote
import stem.util.tor_tools
# bizarre uncatchable stem error
import stem.response.protocolinfo
import stem.response.mapaddress
import stem.response.getconf
import stem.response.getinfo
import stem.response.authchallenge
# 'tuple' object has no attribute 'endswith'
if len(sys.argv) > 1:
lArgs = sys.argv[1:]
LOG.info(f"Getting some {len(lArgs)}")

1
src/stem_examples/compare_flags.py
View File

@ -1,3 +1,4 @@
#!/usr/local/bin/python3.sh
# -*-mode: python; py-indent-offset: 2; indent-tabs-mode: nil; coding: utf-8-unix -*-
#
__doc__ = """

62
src/stem_examples/exit_used.py
View File

@ -1,3 +1,4 @@
#!/usr/local/bin/python3.sh
# -*-mode: python; py-indent-offset: 2; indent-tabs-mode: nil; coding: utf-8-unix -*-
# https://stem.torproject.org/tutorials/examples/exit_used.html
__doc__ = """Determine The Exit You're Using
@ -11,42 +12,69 @@ How can you figure out what exit you're using?
import functools
import sys
import os
import logging
from stem import StreamStatus
from stem.control import EventType, Controller
from stem_examples.tor_controller import get_controller
global COUNT, IMAX
COUNT=0
IMAX = 0
def stream_event(controller, event):
global COUNT, IMAX
COUNT += 1
if IMAX and COUNT >= IMAX:
LOG.info(f"exiting COUNT={COUNT}")
sys.exit(0)
if event.status == StreamStatus.SUCCEEDED and event.circ_id:
circ = controller.get_circuit(event.circ_id)
exit_fingerprint = circ.path[-1][0]
exit_relay = controller.get_network_status(exit_fingerprint)
print("Exit relay for our connection to %s" % (event.target))
print(" address: %s:%i" % (exit_relay.address, exit_relay.or_port))
print(" fingerprint: %s" % exit_relay.fingerprint)
print(" nickname: %s" % exit_relay.nickname)
print(" locale: %s" % controller.get_info("ip-to-country/%s" % exit_relay.address, 'unknown'))
print("")
LOG.info("Exit relay for our connection to %s" % (event.target))
LOG.info(" address: %s:%i" % (exit_relay.address, exit_relay.or_port))
LOG.info(" fingerprint: %s" % exit_relay.fingerprint)
LOG.info(" nickname: %s" % exit_relay.nickname)
LOG.info(" locale: %s" % controller.get_info("ip-to-country/%s" % exit_relay.address, 'unknown'))
LOG.info("")
def iMain():
print("Please wait for requests for tor exits. Press 'enter' to end.")
print("")
if os.path.exists('/run/tor/control'):
controller = get_controller(unix='/run/tor/control')
else:
controller = get_controller(port=9051)
password = os.environ.get('TOR_CONTROLLER_PASSWORD')
controller.authenticate(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)
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__':
iMain()
print('Press Enter')
input() # wait for user to press enter
from stem_examples.stem_utils import vsetup_logging
LOG = logging.getLogger()
if len(sys.argv) > 1:
IMAX = int(sys.argv[1])
else:
IMAX = 0
if os.environ.get('DEBUG', ''):
log_level = 10
else:
log_level = 20
vsetup_logging(LOG, log_level)
try:
iMain()
i = 0
except KeyboardInterrupt as e:
i = 0
except Exception as e:
i = 1
sys.exit(i)

2
src/stem_examples/introduction_points.py
View File

@ -47,7 +47,7 @@ def lMain(lArgs=None, timeout=None):
lArgs = lKNOWN_ONIONS
try:
password = os.environ.get('TOR_CONTROLLER_PASSWORD')
if False and os.path.exists('/run/tor/control'):
if os.path.exists('/run/tor/control'):
controller = get_controller(password=password, unix='/run/tor/control')
else:
controller = get_controller(password=password, port=9051)

1
src/stem_examples/list_circuits.py
View File

@ -1,3 +1,4 @@
#!/usr/local/bin/python3.sh
# -*-mode: python; py-indent-offset: 2; indent-tabs-mode: nil; coding: utf-8-unix -*-
# http://vt5hknv6sblkgf22.onion/tutorials/examples/list_circuits.html

12
src/stem_examples/mappaddress.py
View File

@ -1,5 +1,15 @@
#!/usr/local/bin/python3.sh
# -*-mode: python; py-indent-offset: 2; indent-tabs-mode: nil; coding: utf-8-unix -*-
# https://stem.torproject.org/tutorials/examples/exit_used.html
#
__doc__ = """
Mappaddress queries the socks proxy with an onion address and returns the
IP address that it will use for it. the address will be in the block specified
by the VirtualAddrNetworkIPv4 setting of the torrc, e.g.
VirtualAddrNetworkIPv4 172.16.0.0/12
The script takes one argument, an onion address, without the .onion
"""
import functools
import sys

12
src/stem_examples/outdated_relays.py
View File

@ -9,6 +9,7 @@ to relay operators that needed to upgrade.
https://stem.torproject.org/tutorials/examples/outdated_relays.html
"""
import os
import sys
import logging
@ -41,9 +42,16 @@ def iMain():
return 0
if __name__ == '__main__':
LOG.setLevel(logging.INFO)
from stem_examples.stem_utils import vsetup_logging
LOG = logging.getLogger()
if os.environ.get('DEBUG', ''):
log_level = 10
else:
log_level = 20
vsetup_logging(LOG, log_level)
LOG.setLevel(logging.DEBUG)
try:
l = iMain(sys.argv[1:])
l = iMain()
if l: print(l)
i = 0
except KeyboardInterrupt as e:

1
src/stem_examples/relay_connections.py
View File

@ -1,3 +1,4 @@
#!/usr/local/bin/python3.sh
# -*-mode: python; py-indent-offset: 2; indent-tabs-mode: nil; coding: utf-8-unix -*-
# https://stem.torproject.org/tutorials/examples/relay_connections.html

20
src/stem_examples/tor_bootstrap_check.py
View File

@ -38,7 +38,7 @@ def iMain(lArgs=None):
## TODO: parse the messages above.
print(format(bootstrap_status))
LOG.info(format(bootstrap_status))
progress_percent = re.match('.* PROGRESS=([0-9]+).*', bootstrap_status)
exit_code = int(progress_percent.group(1))
@ -48,5 +48,21 @@ def iMain(lArgs=None):
return exit_code
if __name__ == '__main__':
sys.exit( iMain())
from stem_examples.stem_utils import vsetup_logging
LOG = logging.getLogger()
if os.environ.get('DEBUG', ''):
log_level = 10
else:
log_level = 20
vsetup_logging(LOG, log_level)
LOG.setLevel(logging.DEBUG)
try:
l = iMain()
if l: print(l)
i = 0
except KeyboardInterrupt as e:
i = 0
except Exception as e:
i = 1
sys.exit(i)