diff --git a/modules/onionoo.py b/modules/onionoo.py new file mode 100644 index 00000000..77ccb205 --- /dev/null +++ b/modules/onionoo.py @@ -0,0 +1,61 @@ +#--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: + """ + search = event["args"] + relays = _get_relays_details(search) + event["stdout"].write(_format_relay_summary_message(relays, search))