5678fcdcba
Adds a new torrelay command that can be used to search for a Tor relay by fingerprint/nickname and also a new channel setting that allows relay details to be automatically looked up when a 40-char hex fingerprint appears in the channel.
61 lines
2.4 KiB
Python
61 lines
2.4 KiB
Python
#--depends-on commands
|
|
#--depends-on config
|
|
|
|
import re
|
|
from src import EventManager, ModuleManager, utils
|
|
|
|
REGEX_SHA1_HEX = re.compile("[A-Fa-f0-9]{40}")
|
|
URL_ONIONOO_DETAILS = "https://onionoo.torproject.org/details"
|
|
URL_RELAY_SEARCH_SEARCH = "https://metrics.torproject.org/rs.html#search/"
|
|
URL_RELAY_SEARCH_DETAILS = "https://metrics.torproject.org/rs.html#details/"
|
|
|
|
def _get_relays_details(search):
|
|
page = utils.http.request(
|
|
URL_ONIONOO_DETAILS, get_params={"search": search}, json=True)
|
|
if page and "relays" in page.data:
|
|
return page.data["relays"]
|
|
raise utils.EventsResultsError()
|
|
|
|
def _format_relay_summary_message(relays, search):
|
|
if len(relays) > 1:
|
|
raise utils.EventError(f"There were {len(relays)} relays found for "
|
|
f"this query. {URL_RELAY_SEARCH_SEARCH}{search}")
|
|
if not relays:
|
|
raise utils.EventError("There were no relays found for this query.")
|
|
details = relays[0]
|
|
nickname = details["nickname"]
|
|
consensus_weight = details["consensus_weight"]
|
|
flags = " ".join(details["flags"])
|
|
url = URL_RELAY_SEARCH_DETAILS + details["fingerprint"]
|
|
return f"{nickname} - CW: {consensus_weight} [{flags}] {url}"
|
|
|
|
@utils.export("channelset", utils.BoolSetting("auto-torrelay",
|
|
"Disable/Enable automatically getting Tor relay info from fingerprints"))
|
|
class Module(ModuleManager.BaseModule):
|
|
_name = "Onionoo"
|
|
|
|
@utils.hook("command.regex")
|
|
@utils.kwarg("ignore_action", False)
|
|
@utils.kwarg("priority", EventManager.PRIORITY_MONITOR)
|
|
@utils.kwarg("command", "torrelay")
|
|
@utils.kwarg("pattern", REGEX_SHA1_HEX)
|
|
def channel_message(self, event):
|
|
if event["target"].get_setting("auto-torrelay", False):
|
|
event.eat()
|
|
search = event["match"].group(0)
|
|
try:
|
|
relays = _get_relays_details(search)
|
|
event["stdout"].write(
|
|
_format_relay_summary_message(relays, search))
|
|
except utils.EventError:
|
|
pass
|
|
|
|
@utils.hook("received.command.torrelay", min_args=1)
|
|
def torrelay(self, event):
|
|
"""
|
|
:help: Get summary information about a Tor relay
|
|
:usage: <fingerprint|nickname>
|
|
"""
|
|
search = event["args"]
|
|
relays = _get_relays_details(search)
|
|
event["stdout"].write(_format_relay_summary_message(relays, search))
|