Issue #23: Implement some orconn accounting

Needs tests and testing.

src/vanguards/bandguards.py

@@ -52,6 +52,7 @@ def __init__(self, circ_id, is_hs):
     self.is_hs = is_hs
     self.is_service = 1
     self.is_hsdir = 0
+    self.in_use = 0
     self.created_at = time.time()
     self.read_bytes = 0
     self.sent_bytes = 0
@@ -86,17 +87,72 @@ def dropped_read_bytes_extra(self):
   def dropped_read_rate(self):
     return self.dropped_read_bytes_extra()/self.read_bytes
 
+class BwGuardStat:
+  def __init__(self, guard_fp):
+    self.to_guard = guard_fp
+    self.circs_destroyed = 0
+    self.conns_made = 0
+    self.close_reasons = {} # key=reason val=count
+
 class BandwidthStats:
   def __init__(self, controller):
     self.controller = controller
-    self.circs = {}
+    self.circs = {} # key=circid val=BwCircStat
+    self.live_guard_conns = {} # key=connid val=BwGuardStat
+    self.guards = {} # key=guardfp val=BwGuardStat
+    self.circs_destroyed_total = 0
+    self.no_conns_since = time.time()
+    self._orconn_init(controller)
+
+  def _orconn_init(self, controller):
+    # Load in our current orconns
+    fake_id = 0
+    for l in controller.get_info("orconn-status").split("\n"):
+      self.orconn_event(
+       stem.response.ControlMessage.from_str(
+         "650 ORCONN "+l+" ID="+str(fake_id)+"\r\n", "EVENT"))
+      fake_id += 1
+
+  def orconn_event(self, event):
+    guard_fp = event.endpoint_fingerprint
+    if not event.endpoint_fingerprint in self.guards:
+      self.guards[guard_fp] = BwGuardStat(guard_fp)
+
+    if event.status == "CONNECTED":
+      self.live_guard_conns[event.id] = self.guards[guard_fp]
+      self.guards[guard_fp].conns_made += 1
+      self.no_conns_since = 0
+    if event.status == "CLOSED" or event.status == "FAILED":
+      if event.id in self.live_guard_conns:
+        del self.live_guard_conns[event.id]
+        if len(self.live_guard_conns) == 0:
+          self.no_conns_since = time.time()
+      if event.status == "CLOSED":
+        if not event.reason in self.guards[guard_fp].close_reasons:
+          self.guards[guard_fp].close_reasons[event.reason] = 0
+        self.guards[guard_fp].close_reasons[event.reason] += 1
+    plog("NOTICE", event.raw_content())
+
+  def circuit_destroyed(self, event):
+    # XXX: Limit to warn after?
+    self.destroyed_circs += 1
+    guardfp = event.path[0][0]
+    self.guards[guardfp].circs_destroyed += 1
+    self.circs_destroyed_total += 1
+    plog("NOTICE", "The connection to guard "+guard+" was closed with "+\
+         "an in-use circuit on it. This makes %d for the guard and "+\
+         "%d total." % self.guards[guardfp].circs_destroyed,
+         self.circs_destroyed_total)
 
   def circ_event(self, event):
     # Sometimes circuits get multiple FAILED+CLOSED events,
     # so we must check that first...
     if (event.status == stem.CircStatus.FAILED or \
        event.status == stem.CircStatus.CLOSED):
       if event.id in self.circs:
+        if self.circs[event.id].in_use and \
+           event.remote_reason == "CHANNEL_CLOSED":
+          self.circuit_destroyed(event)
         plog("DEBUG", "Closed hs circ for "+event.raw_content())
         del self.circs[event.id]
     elif event.id not in self.circs:
@@ -113,6 +169,12 @@ def circ_event(self, event):
           self.circs[event.id].is_hsdir = 1
         plog("DEBUG", "Added hs circ for "+event.raw_content())
 
+    # Consider all BUILT circs that have a specific HS purpose
+    # to be "in_use".
+    if event.status == stem.CircStatus.BUILT:
+      if event.purpose[0:9] == "HS_CLIENT" or \
+         event.purpose[0:10] == "HS_SERVICE":
+        self.circs[event.id].in_use = 1
 
   # We need CIRC_MINOR to determine client from service as well
   # as recognize cannibalized HSDIR circs
@@ -128,6 +190,11 @@ def circ_minor_event(self, event):
        event.purpose == "HS_SERVICE_HSDIR":
       self.circs[event.id].is_hsdir = 1
 
+    # PURPOSE_CHANGED from HS_VANGUARDS -> in_use
+    if event.event == stem.CircEvent.PURPOSE_CHANGED:
+      if event.old_purpose == "HS_VANGUARDS":
+        self.circs[event.id].in_use = 1
+
     plog("DEBUG", event.raw_content())
 
   def circbw_event(self, event):
@@ -156,13 +223,25 @@ def circbw_event(self, event):
 
       self.check_circuit_limits(self.circs[event.id])
 
-  def bw_event(self, event):
+  def check_connectivity(self, now):
+    # XXX: limits on logs, configs on values
+    if self.no_conns_since:
+      disconnected_secs = int(now - self.no_conns_since)
+
+      if disconnected_secs >= 10:
+        if disconnected_secs % 10 == 0:
+          plog("WARN", "No live guard connectsions for %d seconds!"
+               % disconnected_secs)
+      elif disconnected_secs >= 5:
+        plog("NOTICE", "No live guard connectsions for %d seconds."
+                 % disconnected_secs)
+
+  def check_circ_ages(self, now):
     if CIRC_MAX_AGE_HOURS <= 0:
       return
 
-    now = time.time()
     # Unused except to expire circuits -- 1x/sec
-    # FIXME: This is has needless copying on python 2..
+    # FIXME: This is needless copying on python 2..
     kill_circs = list(filter(
                         lambda c: now - c.created_at > \
                                   CIRC_MAX_AGE_HOURS*_SECS_PER_HOUR,
@@ -174,6 +253,12 @@ def bw_event(self, event):
                           CIRC_MAX_AGE_HOURS)
       control.try_close_circuit(self.controller, circ.circ_id)
 
+  # Used for 1x/sec heartbeat only
+  def bw_event(self, event):
+    now = time.time()
+    self.check_connectivity(now)
+    self.check_circ_ages(now)
+
   def check_circuit_limits(self, circ):
     if not circ.is_hs: return
     if circ.read_bytes > 0 and \

src/vanguards/main.py

@@ -91,9 +91,6 @@ def control_loop(state):
     plog("NOTICE", "Updated vanguards in torrc. Exiting.")
     sys.exit(0)
 
-  timeouts = cbtverify.TimeoutStats()
-  bandwidths = bandguards.BandwidthStats(controller)
-
   # Thread-safety: state, timeouts, and bandwidths are effectively
   # transferred to the event thread here. They must not be used in
   # our thread anymore.
@@ -105,12 +102,18 @@ def control_loop(state):
                                   stem.control.EventType.CIRC)
 
   if config.ENABLE_BANDGUARDS:
+    bandwidths = bandguards.BandwidthStats(controller)
+
     controller.add_event_listener(
                  functools.partial(bandguards.BandwidthStats.circ_event, bandwidths),
                                   stem.control.EventType.CIRC)
     controller.add_event_listener(
                  functools.partial(bandguards.BandwidthStats.bw_event, bandwidths),
                                   stem.control.EventType.BW)
+    controller.add_event_listener(
+                 functools.partial(bandguards.BandwidthStats.orconn_event, bandwidths),
+                                  stem.control.EventType.ORCONN)
+
     if controller.get_version() >= _MIN_TOR_VERSION_FOR_BW:
       controller.add_event_listener(
                    functools.partial(bandguards.BandwidthStats.circbw_event, bandwidths),
@@ -123,6 +126,8 @@ def control_loop(state):
                       "enabled, you must use Tor 0.3.4.4-rc or newer.")
 
   if config.ENABLE_CBTVERIFY:
+    timeouts = cbtverify.TimeoutStats()
+
     controller.add_event_listener(
                  functools.partial(cbtverify.TimeoutStats.circ_event, timeouts),
                                   stem.control.EventType.CIRC)

tests/test_bandguards.py

@@ -39,6 +39,11 @@ def close_circuit(self, circ_id):
     self.bwstats.circ_event(closed_circ(circ_id))
     raise stem.InvalidRequest("Coverage")
 
+  def get_info(self, key):
+    if key == "orconn-status":
+      return "$3E53D3979DB07EFD736661C934A1DED14127B684~Unnamed CONNECTED\n"+\
+             "$3E53D3979DB07EFD736661C934A1DED14127B684~Unnamed CONNECTED"
+
 def built_circ(circ_id, purpose):
   s = "650 CIRC "+str(circ_id)+" BUILT $5416F3E8F80101A133B1970495B04FDBD1C7446B~Unnamed,$1F9544C0A80F1C5D8A5117FBFFB50694469CC7F4~as44194l10501,$DBD67767640197FF96EC6A87684464FC48F611B6~nocabal,$387B065A38E4DAA16D9D41C2964ECBC4B31D30FF~redjohn1 BUILD_FLAGS=IS_INTERNAL,NEED_CAPACITY,NEED_UPTIME PURPOSE="+purpose+" TIME_CREATED=2018-05-04T06:09:32.751920\r\n"
   return ControlMessage.from_str(s, "EVENT")

tests/test_main.py

@@ -64,6 +64,9 @@ def get_conf(self, key):
       return DATA_DIR
 
   def get_info(self, key, default=None):
+    if key == "orconn-status":
+      return "$3E53D3979DB07EFD736661C934A1DED14127B684~Unnamed CONNECTED\n"+\
+             "$3E53D3979DB07EFD736661C934A1DED14127B684~Unnamed CONNECTED"
     return default
 
   def set_conf(self, key, val):