commit 2e16103be4c892d8d02b7cefd4edc9ecd92833c3 Author: Arturo Filastò <art@fuffa.org> Date: Thu Jun 19 15:02:05 2014 +0200 Implement system for keeping track of reports that have not been submitted. --- ooni/director.py | 45 ++++--- ooni/nettest.py | 7 +- ooni/oonicli.py | 17 +-- ooni/reporter.py | 293 ++++++++++++++++++++++++------------------- ooni/tasks.py | 26 +--- ooni/tests/test_nettest.py | 32 +++-- ooni/tests/test_reporter.py | 12 +- 7 files changed, 220 insertions(+), 212 deletions(-) diff --git a/ooni/director.py b/ooni/director.py index 21ac9fc..1414184 100644 --- a/ooni/director.py +++ b/ooni/director.py @@ -13,10 +13,12 @@ from txtorcon import TorConfig, TorState, launch_tor, build_tor_connection from twisted.internet import defer, reactor from twisted.internet.endpoints import TCP4ClientEndpoint + class Director(object): + """ - Singleton object responsible for coordinating the Measurements Manager and the - Reporting Manager. + Singleton object responsible for coordinating the Measurements Manager + and the Reporting Manager. How this all looks like is as follows: @@ -55,6 +57,7 @@ class Director(object): +------+ """ + _scheduledTests = 0 # Only list NetTests belonging to these categories categories = ['blocking', 'manipulation'] @@ -91,9 +94,10 @@ class Director(object): def getNetTests(self): nettests = {} + def is_nettest(filename): return not filename == '__init__.py' \ - and filename.endswith('.py') + and filename.endswith('.py') for category in self.categories: dirname = os.path.join(config.nettest_directory, category) @@ -106,9 +110,11 @@ class Director(object): if nettest['id'] in nettests: log.err("Found a two tests with the same name %s, %s" % - (net_test_file, nettests[nettest['id']]['path'])) + (net_test_file, + nettests[nettest['id']]['path'])) else: - category = dirname.replace(config.nettest_directory, '') + category = dirname.replace(config.nettest_directory, + '') nettests[nettest['id']] = nettest return nettests @@ -214,7 +220,8 @@ class Director(object): self.allTestsDone.callback(None) @defer.inlineCallbacks - def startNetTest(self, net_test_loader, reporters): + def startNetTest(self, net_test_loader, report_filename, + collector_address=None): """ Create the Report for the NetTest and start the report NetTest. @@ -222,15 +229,19 @@ class Director(object): net_test_loader: an instance of :class:ooni.nettest.NetTestLoader """ + if self.allTestsDone.called: self.allTestsDone = defer.Deferred() if config.privacy.includepcap: if not config.reports.pcap: - config.reports.pcap = config.generate_pcap_filename(net_test_loader.testDetails) + config.reports.pcap = config.generate_pcap_filename( + net_test_loader.testDetails + ) self.startSniffing() - report = Report(reporters, self.reportEntryManager) + report = Report(net_test_loader.testDetails, report_filename, + self.reportEntryManager, collector_address) net_test = NetTest(net_test_loader, report) net_test.director = self @@ -255,7 +266,8 @@ class Director(object): config.scapyFactory = ScapyFactory(config.advanced.interface) if os.path.exists(config.reports.pcap): - log.msg("Report PCAP already exists with filename %s" % config.reports.pcap) + log.msg("Report PCAP already exists with filename %s" % + config.reports.pcap) log.msg("Renaming files with such name...") pushFilenameStack(config.reports.pcap) @@ -268,16 +280,16 @@ class Director(object): @defer.inlineCallbacks def getTorState(self): connection = TCP4ClientEndpoint(reactor, '127.0.0.1', - config.tor.control_port) + config.tor.control_port) config.tor_state = yield build_tor_connection(connection) - def startTor(self): """ Starts Tor Launches a Tor with :param: socks_port :param: control_port :param: tor_binary set in ooniprobe.conf """ log.msg("Starting Tor...") + @defer.inlineCallbacks def state_complete(state): config.tor_state = state @@ -328,9 +340,10 @@ class Director(object): if config.tor.bridges: tor_config.UseBridges = 1 if config.advanced.obfsproxy_binary: - tor_config.ClientTransportPlugin = \ - 'obfs2,obfs3 exec %s managed' % \ - config.advanced.obfsproxy_binary + tor_config.ClientTransportPlugin = ( + 'obfs2,obfs3 exec %s managed' % + config.advanced.obfsproxy_binary + ) bridges = [] with open(config.tor.bridges) as f: for bridge in f: @@ -347,12 +360,12 @@ class Director(object): tor_config.save() - if not hasattr(tor_config,'ControlPort'): + if not hasattr(tor_config, 'ControlPort'): control_port = int(randomFreePort()) tor_config.ControlPort = control_port config.tor.control_port = control_port - if not hasattr(tor_config,'SocksPort'): + if not hasattr(tor_config, 'SocksPort'): socks_port = int(randomFreePort()) tor_config.SocksPort = socks_port config.tor.socks_port = socks_port diff --git a/ooni/nettest.py b/ooni/nettest.py index 4598871..166a4f9 100644 --- a/ooni/nettest.py +++ b/ooni/nettest.py @@ -3,14 +3,12 @@ import re import time from hashlib import sha256 -from twisted.internet import defer, reactor +from twisted.internet import defer from twisted.trial.runner import filenameToModule from twisted.python import usage, reflect -from ooni import geoip from ooni.tasks import Measurement from ooni.utils import log, checkForRoot -from ooni import otime from ooni.settings import config from ooni import errors as e @@ -458,9 +456,6 @@ class NetTest(object): """ self.state.taskDone() - if len(self.report.reporters) == 0: - raise e.AllReportersFailed - return report_results def makeMeasurement(self, test_instance, test_method, test_input=None): diff --git a/ooni/oonicli.py b/ooni/oonicli.py index efcc7db..cc23ec5 100644 --- a/ooni/oonicli.py +++ b/ooni/oonicli.py @@ -13,7 +13,6 @@ from ooni import errors, __version__ from ooni.settings import config from ooni.director import Director from ooni.deck import Deck, nettest_to_path -from ooni.reporter import YAMLReporter, OONIBReporter from ooni.nettest import NetTestLoader from ooni.utils import log, checkForRoot @@ -280,19 +279,9 @@ def runWithDirector(logging=True, start_tor=True): test_details = net_test_loader.testDetails test_details['annotations'] = global_options['annotations'] - yaml_reporter = YAMLReporter(test_details, - report_filename=global_options['reportfile']) - reporters = [yaml_reporter] - - if collector: - log.msg("Reporting using collector: %s" % collector) - try: - oonib_reporter = OONIBReporter(test_details, collector) - reporters.append(oonib_reporter) - except errors.InvalidOONIBCollectorAddress, e: - raise e - - netTestDone = director.startNetTest(net_test_loader, reporters) + director.startNetTest(net_test_loader, + global_options['reportfile'], + collector) return director.allTestsDone def start(): diff --git a/ooni/reporter.py b/ooni/reporter.py index 7d7d709..9869837 100644 --- a/ooni/reporter.py +++ b/ooni/reporter.py @@ -35,7 +35,7 @@ from ooni.utils.net import BodyReceiver, StringProducer from ooni.settings import config -from ooni.tasks import ReportEntry, ReportTracker +from ooni.tasks import ReportEntry def createPacketReport(packet_list): @@ -189,7 +189,7 @@ class YAMLReporter(OReporter): log.msg("Report already exists with filename %s" % report_path) pushFilenameStack(report_path) - self.report_path = report_path + self.report_path = os.path.abspath(report_path) OReporter.__init__(self, test_details) def _writeln(self, line): @@ -396,6 +396,7 @@ class OONIBReporter(OReporter): self.reportID = parsed_response['report_id'] self.backendVersion = parsed_response['backend_version'] log.debug("Created report with id %s" % parsed_response['report_id']) + defer.returnValue(parsed_response['report_id']) @defer.inlineCallbacks def finish(self): @@ -406,26 +407,56 @@ class OONIBReporter(OReporter): class OONIBReportLog(object): + """ + Used to keep track of report creation on a collector backend. + """ + def __init__(self, file_name=config.report_log_file): - self._lock = defer.DeferredLock() self.file_name = file_name + self.create_report_log() + + def run(self, f, *arg, **kw): + lock = defer.DeferredFilesystemLock(self.file_name + '.lock') + d = lock.deferUntilLocked() + + def unlockAndReturn(r): + lock.unlock() + return r + + def execute(_): + d = defer.maybeDeferred(f, *arg, **kw) + d.addBoth(unlockAndReturn) + return d + + d.addCallback(execute) + return d def create_report_log(self): - if os.path.exists(self.file_name): - raise errors.ReportLogExists - with open(self.file_name, 'w+') as f: - f.write(yaml.safe_dump({})) + if not os.path.exists(self.file_name): + with open(self.file_name, 'w+') as f: + f.write(yaml.safe_dump({})) @contextmanager - def edit_report_log(self): + def edit_log(self): with open(self.file_name) as rfp: report = yaml.safe_load(rfp) with open(self.file_name, 'w+') as wfp: yield report wfp.write(yaml.safe_dump(report)) - def _report_created(self, report_file, collector_address, report_id): - with self.edit_report_log() as report: + def _not_created(self, report_file): + with self.edit_log() as report: + report[report_file] = { + 'created_at': datetime.now(), + 'status': 'not-created', + 'collector': None + } + + def not_created(self, report_file): + return self.run(self._not_created, report_file) + + def _created(self, report_file, collector_address, report_id): + with self.edit_log() as report: report[report_file] = { 'created_at': datetime.now(), 'status': 'created', @@ -433,35 +464,45 @@ class OONIBReportLog(object): 'report_id': report_id } - def report_created(self, report_file, collector_address, report_id): - return self._lock.run(self._report_created, report_file, - collector_address, report_id) + def created(self, report_file, collector_address, report_id): + return self.run(self._created, report_file, + collector_address, report_id) - def _report_creation_failed(self, report_file, collector_address): - with self.edit_report_log() as report: + def _creation_failed(self, report_file, collector_address): + with self.edit_log() as report: report[report_file] = { 'created_at': datetime.now(), 'status': 'creation-failed', 'collector': collector_address } - def report_creation_failed(self, report_file, collector_address): - return self._lock.run(self._report_creation_failed, report_file, - collector_address) + def creation_failed(self, report_file, collector_address): + return self.run(self._creation_failed, report_file, + collector_address) - def _report_closed(self, report_file): - with self.edit_report_log() as report: + def _incomplete(self, report_file): + with self.edit_log() as report: + if report[report_file]['status'] != "created": + raise errors.ReportNotCreated() + report[report_file]['status'] = 'incomplete' + + def incomplete(self, report_file): + return self.run(self._incomplete, report_file) + + def _closed(self, report_file): + with self.edit_log() as report: if report[report_file]['status'] != "created": raise errors.ReportNotCreated() del report[report_file] - def report_closed(self, report_file): - return self._lock.run(self._report_closed, report_file) + def closed(self, report_file): + return self.run(self._closed, report_file) class Report(object): - def __init__(self, reporters, reportEntryManager): + def __init__(self, test_details, report_filename, + reportEntryManager, collector_address=None): """ This is an abstraction layer on top of all the configured reporters. @@ -469,53 +510,79 @@ class Report(object): Args: - reporters: - a list of :class:ooni.reporter.OReporter instances + test_details: + A dictionary containing the test details. + + report_filename: + The file path for the report to be written. reportEntryManager: an instance of :class:ooni.tasks.ReportEntryManager + + collector: + The address of the oonib collector for this report. + """ - self.reporters = reporters + self.test_details = test_details + self.collector_address = collector_address + + self.report_log = OONIBReportLog() + + self.yaml_reporter = YAMLReporter(test_details, + report_filename=report_filename) + self.report_filename = self.yaml_reporter.report_path + + self.oonib_reporter = None + if collector_address: + self.oonib_reporter = OONIBReporter(test_details, + collector_address) self.done = defer.Deferred() self.reportEntryManager = reportEntryManager - self._reporters_openned = 0 - self._reporters_written = 0 - self._reporters_closed = 0 + def open_oonib_reporter(self): + def creation_failed(failure): + self.oonib_reporter = None + return self.report_log.creation_failed(self.report_filename) + + def created(report_id): + return self.report_log.created(self.report_filename, + self.collector_address, + report_id) + + d = self.oonib_reporter.createReport() + d.addErrback(creation_failed) + d.addCallback(created) + return d def open(self): """ This will create all the reports that need to be created and fires the created callback of the reporter whose report got created. """ - all_openned = defer.Deferred() + d = defer.Deferred() + deferreds = [] - def are_all_openned(): - if len(self.reporters) == self._reporters_openned: - all_openned.callback(self._reporters_openned) + def yaml_report_failed(failure): + d.errback(failure) - for reporter in self.reporters[:]: + def all_reports_openned(result): + if not d.called: + d.callback(None) - def report_created(result): - log.debug("Created report with %s" % reporter) - self._reporters_openned += 1 - are_all_openned() + if self.oonib_reporter: + deferreds.append(self.open_oonib_reporter()) + else: + deferreds.append(self.report_log.not_created(self.report_filename)) - def report_failed(failure): - try: - self.failedOpeningReport(failure, reporter) - except errors.NoMoreReporters, e: - all_openned.errback(defer.fail(e)) - else: - are_all_openned() - return + yaml_report_created = \ + defer.maybeDeferred(self.yaml_reporter.createReport) + yaml_report_created.addErrback(yaml_report_failed) - d = defer.maybeDeferred(reporter.createReport) - d.addCallback(report_created) - d.addErrback(report_failed) + dl = defer.DeferredList(deferreds) + dl.addCallback(all_reports_openned) - return all_openned + return d def write(self, measurement): """ @@ -532,75 +599,34 @@ class Report(object): been written or errbacks when no more reporters """ - all_written = defer.Deferred() - report_tracker = ReportTracker(self.reporters) - - for reporter in self.reporters[:]: - def report_completed(task): - report_tracker.completed() - if report_tracker.finished(): - all_written.callback(report_tracker) - - def report_failed(failure): - log.debug("Report Write Failure") - try: - report_tracker.failedReporters.append(reporter) - self.failedWritingReport(failure, reporter) - except errors.NoMoreReporters, e: - log.err("No More Reporters!") - all_written.errback(defer.fail(e)) - else: - report_tracker.completed() - if report_tracker.finished(): - all_written.callback(report_tracker) - return - - report_entry_task = ReportEntry(reporter, measurement) - self.reportEntryManager.schedule(report_entry_task) - - report_entry_task.done.addCallback(report_completed) - report_entry_task.done.addErrback(report_failed) - - return all_written - - def failedWritingReport(self, failure, reporter): - """ - This errback gets called every time we fail to write a report. - By fail we mean that the number of retries has exceeded. - Once a report has failed to be written with a reporter we give up and - remove the reporter from the list of reporters to write to. - """ + d = defer.Deferred() + deferreds = [] - # XXX: may have been removed already by another failure. - if reporter in self.reporters: - log.err("Failed to write to %s reporter, giving up..." % reporter) - self.reporters.remove(reporter) - else: - log.err("Failed to write to (already) removed reporter %s" % - reporter) + def yaml_report_failed(failure): + d.errback(failure) - # Don't forward the exception unless there are no more reporters - if len(self.reporters) == 0: - log.err("Removed last reporter %s" % reporter) - raise errors.NoMoreReporters - return + def oonib_report_failed(failure): + return self.report_log.incomplete(self.report_filename) - def failedOpeningReport(self, failure, reporter): - """ - This errback get's called every time we fail to create a report. - By fail we mean that the number of retries has exceeded. - Once a report has failed to be created with a reporter we give up and - remove the reporter from the list of reporters to write to. - """ - log.err("Failed to open %s reporter, giving up..." % reporter) - log.err("Reporter %s failed, removing from report..." % reporter) - if reporter in self.reporters: - self.reporters.remove(reporter) - # Don't forward the exception unless there are no more reporters - if len(self.reporters) == 0: - log.err("Removed last reporter %s" % reporter) - raise errors.NoMoreReporters - return + def all_reports_written(_): + if not d.called: + d.callback(None) + + write_yaml_report = ReportEntry(self.yaml_reporter, measurement) + self.reportEntryManager.schedule(write_yaml_report) + write_yaml_report.done.addErrback(yaml_report_failed) + deferreds.append(write_yaml_report.done) + + if self.oonib_reporter: + write_oonib_report = ReportEntry(self.oonib_reporter, measurement) + self.reportEntryManager.schedule(write_oonib_report) + write_oonib_report.done.addErrback(oonib_report_failed) + deferreds.append(write_oonib_report.done) + + dl = defer.DeferredList(deferreds) + dl.addCallback(all_reports_written) + + return d def close(self): """ @@ -611,20 +637,29 @@ class Report(object): all the reports have been closed. """ - all_closed = defer.Deferred() + d = defer.Deferred() + deferreds = [] + + def yaml_report_failed(failure): + d.errback(failure) + + def oonib_report_closed(result): + return self.report_log.closed(self.report_filename) + + def all_reports_closed(_): + if not d.called: + d.callback(None) - for reporter in self.reporters[:]: - def report_closed(result): - self._reporters_closed += 1 - if len(self.reporters) == self._reporters_closed: - all_closed.callback(self._reporters_closed) + close_yaml = defer.maybeDeferred(self.yaml_reporter.finish) + close_yaml.addErrback(yaml_report_failed) + deferreds.append(close_yaml) - def report_failed(failure): - log.err("Failed closing report") - log.exception(failure) + if self.oonib_reporter: + close_oonib = self.oonib_reporter.finish() + close_oonib.addCallback(oonib_report_closed) + deferreds.append(close_oonib) - d = defer.maybeDeferred(reporter.finish) - d.addCallback(report_closed) - d.addErrback(report_failed) + dl = defer.DeferredList(deferreds) + dl.addCallback(all_reports_closed) - return all_closed + return d diff --git a/ooni/tasks.py b/ooni/tasks.py index 578d8a9..a869254 100644 --- a/ooni/tasks.py +++ b/ooni/tasks.py @@ -126,34 +126,12 @@ class Measurement(TaskWithTimeout): def run(self): return self.netTestMethod() -class ReportTracker(object): - def __init__(self, reporters): - self.report_completed = 0 - self.reporters = reporters - self.failedReporters = [] - - def finished(self): - """ - Returns true if all the tasks are done. False if not. - """ - # If a reporter fails and is removed, the report - # is considered completed but failed, but the number - # of reporters is now decreased by the number of failed - # reporters. - if self.report_completed == (len(self.reporters) + len(self.failedReporters)): - return True - return False - - def completed(self): - """ - Called when a new report is completed. - """ - self.report_completed += 1 class ReportEntry(TaskWithTimeout): + def __init__(self, reporter, entry): self.reporter = reporter - self.entry = entry + self.entry = entry if config.advanced.reporting_timeout: self.timeout = config.advanced.reporting_timeout diff --git a/ooni/tests/test_nettest.py b/ooni/tests/test_nettest.py index 92278ce..6a550d6 100644 --- a/ooni/tests/test_nettest.py +++ b/ooni/tests/test_nettest.py @@ -1,22 +1,15 @@ import os -from StringIO import StringIO -from tempfile import TemporaryFile, mkstemp +from tempfile import mkstemp from twisted.trial import unittest from twisted.internet import defer, reactor from twisted.python.usage import UsageError from ooni.settings import config -from ooni.errors import MissingRequiredOption, InvalidOption, FailureToLoadNetTest +from ooni.errors import MissingRequiredOption from ooni.nettest import NetTest, NetTestLoader -from ooni.tasks import BaseTask from ooni.director import Director -from ooni.managers import TaskManager - -from ooni.tests.mocks import MockMeasurement, MockMeasurementFailOnce -from ooni.tests.mocks import MockNetTest, MockDirector, MockReporter -from ooni.tests.mocks import MockMeasurementManager net_test_string = """ from twisted.python import usage @@ -95,7 +88,6 @@ from ooni.errors import failureToString, handleAllFailures class UsageOptions(usage.Options): optParameters = [ ['url', 'u', None, 'Specify a single URL to test.'], - ['factor', 'f', 0.8, 'What factor should be used for triggering censorship (0.8 == 80%)'] ] class HTTPBasedTest(httpt.HTTPTest): @@ -107,15 +99,17 @@ class HTTPBasedTest(httpt.HTTPTest): dummyInputs = range(1) dummyArgs = ('--spam', 'notham') -dummyOptions = {'spam':'notham'} +dummyOptions = {'spam': 'notham'} dummyInvalidArgs = ('--cram', 'jam') -dummyInvalidOptions= {'cram':'jam'} +dummyInvalidOptions = {'cram': 'jam'} dummyArgsWithRequiredOptions = ('--foo', 'moo', '--bar', 'baz') -dummyRequiredOptions = {'foo':'moo', 'bar':'baz'} +dummyRequiredOptions = {'foo': 'moo', 'bar': 'baz'} dummyArgsWithFile = ('--spam', 'notham', '--file', 'dummyInputFile.txt') + class TestNetTest(unittest.TestCase): timeout = 1 + def setUp(self): with open('dummyInputFile.txt', 'w') as f: for i in range(10): @@ -211,7 +205,7 @@ class TestNetTest(unittest.TestCase): ntl.loadNetTestString(net_test_string_with_file) ntl.checkOptions() - nt = NetTest(ntl,None) + nt = NetTest(ntl, None) nt.initializeInputProcessor() # XXX: if you use the same test_class twice you will have consumed all @@ -249,7 +243,7 @@ class TestNetTest(unittest.TestCase): ntl.checkOptions() director = Director() - d = director.startNetTest(ntl, [MockReporter()]) + d = director.startNetTest(ntl, 'dummy_report.yaml') @d.addCallback def complete(result): @@ -259,24 +253,28 @@ class TestNetTest(unittest.TestCase): return d def test_require_root_succeed(self): - #XXX: will require root to run + # XXX: will require root to run ntl = NetTestLoader(dummyArgs) ntl.loadNetTestString(net_test_root_required) for test_class, method in ntl.testCases: self.assertTrue(test_class.requiresRoot) + class TestNettestTimeout(unittest.TestCase): + @defer.inlineCallbacks def setUp(self): from twisted.internet.protocol import Protocol, Factory from twisted.internet.endpoints import TCP4ServerEndpoint class DummyProtocol(Protocol): + def dataReceived(self, data): pass class DummyFactory(Factory): + def __init__(self): self.protocols = [] @@ -306,7 +304,7 @@ class TestNettestTimeout(unittest.TestCase): ntl.checkOptions() director = Director() - d = director.startNetTest(ntl, [MockReporter()]) + d = director.startNetTest(ntl, 'dummy_report.yaml') @d.addCallback def complete(result): diff --git a/ooni/tests/test_reporter.py b/ooni/tests/test_reporter.py index 5ca7bfa..69d9775 100644 --- a/ooni/tests/test_reporter.py +++ b/ooni/tests/test_reporter.py @@ -111,7 +111,7 @@ class TestOONIBReportLog(unittest.TestCase): @defer.inlineCallbacks def test_report_created(self): - yield self.report_log.report_created("path_to_my_report.yaml", + yield self.report_log.created("path_to_my_report.yaml", 'httpo://foo.onion', 'someid') with open(self.report_log.file_name) as f: @@ -120,10 +120,10 @@ class TestOONIBReportLog(unittest.TestCase): @defer.inlineCallbacks def test_concurrent_edit(self): - d1 = self.report_log.report_created("path_to_my_report1.yaml", + d1 = self.report_log.created("path_to_my_report1.yaml", 'httpo://foo.onion', 'someid1') - d2 = self.report_log.report_created("path_to_my_report2.yaml", + d2 = self.report_log.created("path_to_my_report2.yaml", 'httpo://foo.onion', 'someid2') yield defer.DeferredList([d1, d2]) @@ -134,10 +134,10 @@ class TestOONIBReportLog(unittest.TestCase): @defer.inlineCallbacks def test_report_closed(self): - yield self.report_log.report_created("path_to_my_report.yaml", + yield self.report_log.created("path_to_my_report.yaml", 'httpo://foo.onion', 'someid') - yield self.report_log.report_closed("path_to_my_report.yaml") + yield self.report_log.closed("path_to_my_report.yaml") with open(self.report_log.file_name) as f: report = yaml.safe_load(f) @@ -145,7 +145,7 @@ class TestOONIBReportLog(unittest.TestCase): @defer.inlineCallbacks def test_report_creation_failed(self): - yield self.report_log.report_creation_failed("path_to_my_report.yaml", + yield self.report_log.creation_failed("path_to_my_report.yaml", 'httpo://foo.onion') with open(self.report_log.file_name) as f: report = yaml.safe_load(f)