#!/usr/bin/env python3
import os
import json
import logging
import pprint
import argparse
import datetime
try:
import requests
except ImportError:
print("Error: requests module is required. apt install python3-requests")
exit(1)
# Disabling annoying warnings
requests.packages.urllib3.disable_warnings(
requests.packages.urllib3.exceptions.InsecureRequestWarning)
INFO_METHODS = {
"RouterInfo": {
"i2p.router.uptime": "",
"i2p.router.net.status": "",
"i2p.router.netdb.knownpeers": "",
"i2p.router.netdb.activepeers": "",
"i2p.router.net.bw.inbound.1s": "",
"i2p.router.net.bw.outbound.1s": "",
"i2p.router.net.tunnels.participating": "",
"i2p.router.net.tunnels.successrate": "",
"i2p.router.net.total.received.bytes": "",
"i2p.router.net.total.sent.bytes": "",
},
"ClientServicesInfo": {
"I2PTunnel": "",
"SOCKS": "",
"HTTPProxy": "",
"SAM": "",
"BOB": "",
"I2CP": "",
}
}
STATUS = [
"OK",
"TESTING",
"FIREWALLED",
"HIDDEN",
"WARN_FIREWALLED_AND_FAST",
"WARN_FIREWALLED_AND_FLOODFILL",
"WARN_FIREWALLED_WITH_INBOUND_TCP",
"WARN_FIREWALLED_WITH_UDP_DISABLED",
"ERROR_I2CP",
"ERROR_CLOCK_SKEW",
"ERROR_PRIVATE_TCP_ADDRESS",
"ERROR_SYMMETRIC_NAT",
"ERROR_UDP_PORT_IN_USE",
"ERROR_NO_ACTIVE_PEERS_CHECK_CONNECTION_AND_FIREWALL",
"ERROR_UDP_DISABLED_AND_TCP_UNSET",
]
SUFFIXES = ['B','KB','MB','GB','TB']
def humanize_size(size, precision=2):
"""Bytes to human readable size"""
suffixIndex = 0
while size > 1024:
suffixIndex += 1
size = size / 1024.0
return "{}{}".format(round(size ,precision), SUFFIXES[suffixIndex])
def humanize_time(milliseconds):
"""Seconds to human readable time"""
return str(datetime.timedelta(milliseconds=milliseconds))
class I2PControl(object):
"""Talk to I2PControl API"""
def __init__(self, url, password='itoopie'):
self.url = url
self.password = password
self._token = None
@property
def token(self):
"""Cached authentication token"""
if not self._token:
self._token = requests.post(self.url,
json.dumps({'id': 1, 'method': 'Authenticate',
'params': {'API': 1, 'Password': self.password},
'jsonrpc': '2.0'}),
verify=False).json()["result"]["Token"]
return self._token
def request(self, method, params):
"""Execute authenticated request"""
return requests.post(self.url,
json.dumps({'id': 1, 'method': method, 'params': params,
'jsonrpc': '2.0', 'Token': self.token}),
verify=False
)
class I2pdctl(object):
"""i2pd control"""
def __init__(self, ctl):
self.ctl = ctl
def execute(self, args):
"""Execute raw method"""
try:
parameters = json.loads(args.parameters)
except json.decoder.JSONDecodeError:
print("Error: parameters should be a valid JSON object")
return
resp = self.ctl.request(args.method, parameters)
if args.json_output:
print(resp.text)
elif "result" in resp:
pprint.pprint(resp.json()["result"])
else:
pprint.pprint(resp.json())
def raw_info(self, args):
"""Retrieve raw JSON reply from method(s)"""
res = {}
for m in args.method:
if m not in INFO_METHODS:
print("Invalid method {}. Supported methods are {}".format(m,
", ".join(INFO_METHODS.keys())))
return
else:
res[m] = self.ctl.request(m, INFO_METHODS[m]).json()['result']
print(json.dumps(res))
def print_info(self, args):
"""Print information about a node in a human readable form"""
def fancy_title(string):
print("\n### {}".format(string))
ri_res = self.ctl.request("RouterInfo", INFO_METHODS["RouterInfo"]).json()['result']
try:
csi_res = self.ctl.request("ClientServicesInfo", INFO_METHODS["ClientServicesInfo"]).json()['result']
except KeyError:
csi_res = False
fancy_title("Router info")
print("Uptime: {}".format(
humanize_time(int(ri_res["i2p.router.uptime"]))))
print("Status: {}".format(STATUS[ri_res["i2p.router.net.status"]]))
print("Tunnel creation success rate: {}%".format(
ri_res["i2p.router.net.tunnels.successrate"]))
print("Received: {} ({} B/s) / Sent: {} ({} B/s)".format(
humanize_size(int(ri_res["i2p.router.net.total.received.bytes"])),
humanize_size(int(ri_res["i2p.router.net.bw.inbound.1s"])),
humanize_size(int(ri_res["i2p.router.net.total.sent.bytes"])),
humanize_size(int(ri_res["i2p.router.net.bw.outbound.1s"]))))
print("Known routers: {} / Active: {}".format(
ri_res["i2p.router.netdb.knownpeers"],
ri_res["i2p.router.netdb.activepeers"]))
if csi_res:
fancy_title("Interfaces")
for n in ["HTTPProxy", "SOCKS", "BOB", "SAM", "I2CP"]:
print("- {}:".format(n), "ON" if csi_res[n]["enabled"] == "true" else "OFF")
if csi_res["I2PTunnel"]["client"]:
fancy_title("Client I2P Tunnels")
for tunnel in sorted(list(csi_res["I2PTunnel"]["client"].keys())):
print("-", tunnel, csi_res["I2PTunnel"]["client"][tunnel])
if csi_res["I2PTunnel"]["server"]:
fancy_title("Server I2P Tunnels")
for tunnel in sorted(list(csi_res["I2PTunnel"]["server"].keys())):
t = csi_res["I2PTunnel"]["server"][tunnel]
if "port" in t:
print("-", tunnel, "{}:{}".format(t["address"], t["port"]))
else:
print("-", tunnel, t["address"])
def main():
URL = os.getenv("I2PCONTROL_URL", "https://127.0.0.1:7650/")
PASSWORD = os.getenv("I2PCONTROL_PASSWORD", "itoopie")
ctl = I2pdctl(I2PControl(URL, PASSWORD))
parser = argparse.ArgumentParser()
subparsers = parser.add_subparsers(title="actions",help="Command to execute")
exec_parser = subparsers.add_parser("exec", description="Execute RPC method with parameters, return results or raw JSON")
exec_parser.add_argument('method', help="RPC method name")
exec_parser.add_argument('parameters', help="Parameters as raw JSON string")
exec_parser.add_argument('-j', '--json-output', action="store_true", help="Output raw JSON reply")
exec_parser.set_defaults(func=ctl.execute)
raw_info_parser = subparsers.add_parser("raw_info", description="Retrieve JSON info from specified *Info methods")
raw_info_parser.add_argument('method', nargs='*',
help="RPC method(s) to retreive info. Supported methods: {}".format(", ".join(INFO_METHODS.keys())))
raw_info_parser.set_defaults(func=ctl.raw_info)
print_info_parser = subparsers.add_parser("info", description="Print generic information about node in a human readable form")
print_info_parser.set_defaults(func=ctl.print_info)
args = parser.parse_args()
if hasattr(args, "func"):
try:
args.func(args)
except requests.exceptions.ConnectionError:
print("Error: I2PControl URL is unavailable. Check your i2pd settings and network connection.")
exit(1)
else:
parser.print_help()
if __name__ == "__main__":
main()