diff --git a/docs/topics/development/scanner_pipeline.md b/docs/topics/development/scanner_pipeline.md index 13b4893a8ca5..08d788cb2727 100644 --- a/docs/topics/development/scanner_pipeline.md +++ b/docs/topics/development/scanner_pipeline.md @@ -132,6 +132,12 @@ When some results are still missing two hours after the version was created, the with the same payload, and keeps doing so every two hours, up to 12 times. The last retry therefore happens 24 hours after the version was created. +AMO waits in two steps, each with a budget of its own: first for the results +[`on_version_scanned`](#scanner-on-version-scanned) aggregates, then for +`on_version_scanned` itself, which cannot be sent before they are all in. A +scanner listening to `on_version_scanned` therefore gets the same 24 hours, +counted from the moment the event was sent to it. + Once the retries are exhausted, or when the payload can no longer be rebuilt (e.g., the uploaded file a `during_validation` payload points to is gone), AMO records artificial results matching the special `SCANNER_RESULTS_MISSING` @@ -480,6 +486,70 @@ The payload sent looks like this: } ``` +(scanner-on-version-scanned)= +### `on_version_scanned` + +This event occurs once **every** scanner listening to a predefined set of events +has returned its results for the version. It is meant for scanners that need to +make a decision informed by what the other scanners found. + +In addition to the usual `addon` and `version` properties, the payload contains a +`scanner_results` object. It has one property per aggregated event, each mapping +the name of a scanner listening to that event to the results it returned. A +`null` value means that scanner skipped the event. + +A scanner listening to both aggregated events appears under both, so its two sets +of results remain distinguishable. + +```json +{ + "addon": { + // Similar to the `on_version_created` event. + }, + "version": { + // Similar to the `on_version_created` event. + }, + "scanner_results": { + "during_validation": { + "some-scanner": { + "version": "1.0", + "matchedRules": ["SOME_RULE"], + "annotations": {} + } + }, + "on_version_created": { + "some-scanner": { + "version": "1.0", + "matchedRules": [], + "annotations": {} + }, + "some-other-scanner": null + } + }, + "event": "on_version_scanned", + "scanner_result_url": "http://olympia.test/api/v5/scanner/results/127/" +} +``` + +```{warning} +`scanner_results` reports every result AMO holds for an aggregated event, which +is not the same set as the scanners currently subscribed to it. A scanner that +subscribed *after* the version was created is missing entirely, because AMO +never called it; conversely, a scanner whose webhook has since been deactivated +is still reported, because its results had already been collected. Don't assume +the object lists every scanner currently subscribed; look up the scanners you +care about by name and handle their absence. +``` + +This event blocks auto-approval like the events it aggregates, so the version is +not auto-approved until the scanners listening to it have answered as well. When +AMO gave up on one of the aggregated scanners, the event is still sent, with the +`SCANNER_RESULTS_MISSING` [rule](#scanner-rules) in place of that scanner's +results. + +Results [pushed](#scanner-push) by a scanner are not part of the payload, and +pushing results does not trigger this event. + (scanner-push)= ### `push` diff --git a/src/olympia/constants/scanners.py b/src/olympia/constants/scanners.py index 9efbec5f9e65..fce8748b9932 100644 --- a/src/olympia/constants/scanners.py +++ b/src/olympia/constants/scanners.py @@ -81,6 +81,7 @@ WEBHOOK_ON_VERSION_CREATED = 3 WEBHOOK_PUSH = 4 WEBHOOK_ON_LISTING_CHANGED = 5 +WEBHOOK_ON_VERSION_SCANNED = 6 WEBHOOK_EVENTS = { WEBHOOK_DURING_VALIDATION: 'during_validation', @@ -88,14 +89,22 @@ WEBHOOK_ON_VERSION_CREATED: 'on_version_created', WEBHOOK_PUSH: 'push', WEBHOOK_ON_LISTING_CHANGED: 'on_listing_changed', + WEBHOOK_ON_VERSION_SCANNED: 'on_version_scanned', } -# Events we wait on before auto-approving a version. -WEBHOOK_EVENTS_BLOCKING_AUTO_APPROVAL = [ +# The events whose results are aggregated into the `on_version_scanned` +# payload. We wait for all of them before sending it. +WEBHOOK_EVENTS_AGGREGATED_BY_ON_VERSION_SCANNED = [ WEBHOOK_DURING_VALIDATION, WEBHOOK_ON_VERSION_CREATED, ] +# Events we wait on before auto-approving a version. +WEBHOOK_EVENTS_BLOCKING_AUTO_APPROVAL = [ + *WEBHOOK_EVENTS_AGGREGATED_BY_ON_VERSION_SCANNED, + WEBHOOK_ON_VERSION_SCANNED, +] + # Number of retries before we give up on the scanners for a version. WEBHOOK_MAX_RETRIES = 12 diff --git a/src/olympia/lib/settings_base.py b/src/olympia/lib/settings_base.py index 19485a77ddf5..1f587c03865d 100644 --- a/src/olympia/lib/settings_base.py +++ b/src/olympia/lib/settings_base.py @@ -832,6 +832,7 @@ def get_language_url_map(): 'olympia.files.tasks.repack_fileupload': {'queue': 'devhub'}, 'olympia.scanners.tasks.call_webhooks_during_validation': {'queue': 'devhub'}, 'olympia.scanners.tasks.call_webhooks_on_listing_changed': {'queue': 'devhub'}, + 'olympia.scanners.tasks.call_webhooks_on_version_scanned': {'queue': 'devhub'}, 'olympia.scanners.tasks.run_actions_for_scanner_result': {'queue': 'devhub'}, 'olympia.scanners.tasks.run_narc_on_version': {'queue': 'devhub'}, 'olympia.scanners.tasks.run_yara': {'queue': 'devhub'}, diff --git a/src/olympia/reviewers/tests/test_models.py b/src/olympia/reviewers/tests/test_models.py index 045d673fb74f..5dca10f1584d 100644 --- a/src/olympia/reviewers/tests/test_models.py +++ b/src/olympia/reviewers/tests/test_models.py @@ -26,7 +26,11 @@ version_factory, ) from olympia.blocklist.models import BlockVersion -from olympia.constants.scanners import WEBHOOK, WEBHOOK_DURING_VALIDATION +from olympia.constants.scanners import ( + WEBHOOK, + WEBHOOK_DURING_VALIDATION, + WEBHOOK_ON_VERSION_SCANNED, +) from olympia.files.models import File, FileValidation, WebextPermission from olympia.promoted.models import ( PromotedAddon, @@ -1505,6 +1509,22 @@ def test_check_is_waiting_on_scanners_result_without_matched_rules(self): ) assert AutoApprovalSummary.check_is_waiting_on_scanners(self.version) is True + def test_check_is_waiting_on_scanners_on_version_scanned_event(self): + webhook = ScannerWebhook.objects.create(name='some-scanner') + webhook.update(modified=self.days_ago(1)) + event = ScannerWebhookEvent.objects.create( + event=WEBHOOK_ON_VERSION_SCANNED, webhook=webhook + ) + assert AutoApprovalSummary.check_is_waiting_on_scanners(self.version) is True + + ScannerResult.objects.create( + version=self.version, + scanner=WEBHOOK, + webhook_event=event, + results={'matchedRules': []}, + ) + assert AutoApprovalSummary.check_is_waiting_on_scanners(self.version) is False + def test_check_is_waiting_on_scanners_multiple_events_all_complete(self): webhook = ScannerWebhook.objects.create(name='some-scanner') webhook.update(modified=self.days_ago(1)) diff --git a/src/olympia/scanners/management/commands/retry_versions_waiting_on_scanners.py b/src/olympia/scanners/management/commands/retry_versions_waiting_on_scanners.py index e6283c948e52..4feeaa42a462 100644 --- a/src/olympia/scanners/management/commands/retry_versions_waiting_on_scanners.py +++ b/src/olympia/scanners/management/commands/retry_versions_waiting_on_scanners.py @@ -4,19 +4,19 @@ from django.core.management.base import BaseCommand import olympia.core.logger -from olympia.constants.scanners import ( - WEBHOOK_EVENTS_BLOCKING_AUTO_APPROVAL, - WEBHOOK_MAX_RETRIES, +from olympia.constants.scanners import WEBHOOK_MAX_RETRIES +from olympia.scanners.tasks import ( + scanner_webhook_events_to_wait_for, + wait_for_scanner_results, ) -from olympia.scanners.tasks import wait_for_scanner_results from olympia.versions.models import Version log = olympia.core.logger.getLogger('z.scanners.retry_versions_waiting_on_scanners') # How long `wait_for_scanner_results` can keep retrying a version: the initial -# countdown plus one delay per retry. -MIN_AGE = timedelta( +# countdown plus one delay per retry, for each of the two steps we wait in. +MIN_AGE = 2 * timedelta( seconds=settings.SCANNER_WEBHOOK_RETRY_DELAY * (WEBHOOK_MAX_RETRIES + 1) ) @@ -64,7 +64,7 @@ def handle(self, *args, **options): wait_for_scanner_results.delay( version_pk=version.pk, - event_ids=WEBHOOK_EVENTS_BLOCKING_AUTO_APPROVAL, + event_ids=scanner_webhook_events_to_wait_for(version), ) log.info( 'Scheduled wait_for_scanner_results again for version %s.', version.pk diff --git a/src/olympia/scanners/migrations/0093_alter_scannerwebhookevent_event.py b/src/olympia/scanners/migrations/0093_alter_scannerwebhookevent_event.py new file mode 100644 index 000000000000..5c42d5d409e9 --- /dev/null +++ b/src/olympia/scanners/migrations/0093_alter_scannerwebhookevent_event.py @@ -0,0 +1,24 @@ +from django.db import migrations, models + + +class Migration(migrations.Migration): + dependencies = [ + ('scanners', '0092_alter_scannerwebhookevent_event'), + ] + + operations = [ + migrations.AlterField( + model_name='scannerwebhookevent', + name='event', + field=models.PositiveSmallIntegerField( + choices=[ + (1, 'during_validation'), + (2, 'on_source_code_uploaded'), + (3, 'on_version_created'), + (4, 'push'), + (5, 'on_listing_changed'), + (6, 'on_version_scanned'), + ] + ), + ), + ] diff --git a/src/olympia/scanners/tasks.py b/src/olympia/scanners/tasks.py index f6b95ce3e722..9615dbdc752f 100644 --- a/src/olympia/scanners/tasks.py +++ b/src/olympia/scanners/tasks.py @@ -7,6 +7,7 @@ from collections import defaultdict from django.conf import settings +from django.db import transaction from django.db.models import F from django.template.defaultfilters import slugify from django.urls import reverse @@ -41,10 +42,11 @@ WEBHOOK, WEBHOOK_DURING_VALIDATION, WEBHOOK_EVENTS, - WEBHOOK_EVENTS_BLOCKING_AUTO_APPROVAL, + WEBHOOK_EVENTS_AGGREGATED_BY_ON_VERSION_SCANNED, WEBHOOK_MAX_RETRIES, WEBHOOK_ON_LISTING_CHANGED, WEBHOOK_ON_VERSION_CREATED, + WEBHOOK_ON_VERSION_SCANNED, YARA, ) from olympia.devhub.tasks import validation_task @@ -172,14 +174,25 @@ def _deliver_webhook(*, scanner_result, payload, request_id): ) scanner_result.reload() - if scanner_result.results != previous_results: - # The scanner sent its results while we were calling it. - return + if scanner_result.results == previous_results: + # Don't overwrite the results the scanner may have sent while we were + # calling it. + scanner_result.results = data + # We don't pass `update_fields` because the `save()` method also + # updates other fields (e.g. has_matches, matched_rules). + scanner_result.save() - scanner_result.results = data - # We don't pass `update_fields` because the `save()` method - # also updates other fields (e.g. has_matches, matched_rules). - scanner_result.save() + maybe_call_webhooks_on_version_scanned(scanner_result) + + +def maybe_call_webhooks_on_version_scanned(scanner_result): + if ( + scanner_result.version_id + and scanner_result.is_complete + and scanner_result.webhook_event.event + in WEBHOOK_EVENTS_AGGREGATED_BY_ON_VERSION_SCANNED + ): + call_webhooks_on_version_scanned.delay(version_pk=scanner_result.version_id) @task @@ -232,9 +245,45 @@ def build_webhook_payload(event_id, *, upload=None, version=None): 'version': WebhookVersionSerializer(version).data, } + if event_id == WEBHOOK_ON_VERSION_SCANNED: + return { + 'addon': WebhookAddonSerializer(version.addon).data, + 'version': WebhookVersionSerializer(version).data, + 'scanner_results': _build_scanner_results(version), + } + raise ValueError(f'No payload for webhook event {event_id}') +def _build_scanner_results(version): + """Return the results of the events aggregated into `on_version_scanned`, + as a mapping of event name to scanner name to results.""" + # Always expose every aggregated event, even when no scanner is listening + # to it, so that the payload has a stable shape. + scanner_results = { + WEBHOOK_EVENTS[event_id]: {} + for event_id in WEBHOOK_EVENTS_AGGREGATED_BY_ON_VERSION_SCANNED + } + + # Only completed results, including those of webhooks deactivated since + # the version was created. + for result in ( + ScannerResult.objects.filter( + ScannerResult.complete_q, + version=version, + webhook_event__event__in=WEBHOOK_EVENTS_AGGREGATED_BY_ON_VERSION_SCANNED, + ) + .select_related('webhook_event__webhook') + .order_by('pk') + ): + # When a scanner has more than one result for this event and this + # version, the last one (by pk) wins. + event_name = WEBHOOK_EVENTS[result.webhook_event.event] + scanner_results[event_name][result.webhook_event.webhook.name] = result.results + + return scanner_results + + def _build_payload_for_result(scanner_result): """Return the payload to send again for an existing scanner result, or None when it cannot be rebuilt.""" @@ -270,6 +319,110 @@ def _record_missing_results(scanner_result): scanner_result.pk, ) + # This is the last thing we do for this version, so this is also our last + # chance to send `on_version_scanned`: the results it aggregates are now + # complete, even though we made some of them up. + maybe_call_webhooks_on_version_scanned(scanner_result) + + +@task +@use_primary_db +def call_webhooks_on_version_scanned(version_pk): + """Call the webhooks subscribed to `on_version_scanned` for this version, + once we have the results the event aggregates. + + Safe to call as often as we want: it does nothing until the version is + ready, and nothing again once the results exist. Creating them is what + makes this run the one that delivers the event, so it is also the run that + starts waiting for the answers.""" + version = Version.unfiltered.get(pk=version_pk) + + if is_waiting_on_scanner_webhook_events( + version=version, event_ids=WEBHOOK_EVENTS_AGGREGATED_BY_ON_VERSION_SCANNED + ): + return + + # The results of the scanners subscribed to `on_version_scanned`, which + # they will fill in. Not to be confused with the results we send them. + pending_results = [] + with transaction.atomic(): + # Lock the version row, so that two runs cannot create two results for + # the same event and deliver it twice. `exists()` drops the manager's + # `select_related()`, so we only lock that row. + Version.unfiltered.select_for_update().filter(pk=version_pk).exists() + + for event in ScannerWebhookEvent.to_wait_for( + version=version, event_ids=[WEBHOOK_ON_VERSION_SCANNED] + ): + if ScannerResult.objects.filter( + version=version, webhook_event=event + ).exists(): + # We have already delivered this event. + continue + + pending_results.append( + ScannerResult.objects.create( + scanner=WEBHOOK, webhook_event=event, version=version + ) + ) + + if not pending_results: + return + + # Built once: every scanner gets the same results. + payload = build_webhook_payload(WEBHOOK_ON_VERSION_SCANNED, version=version) + + for scanner_result in pending_results: + event = scanner_result.webhook_event + statsd_name = _get_webhook_statsd_name(event) + request_id = uuid.uuid4().hex + log.info( + 'Calling webhook "%s" for version %s request_id=%s.', + event.webhook.name, + version_pk, + request_id, + ) + + try: + _deliver_webhook( + scanner_result=scanner_result, + payload=payload, + request_id=request_id, + ) + statsd.incr(f'{statsd_name}.success') + except Exception: + statsd.incr(f'{statsd_name}.failure') + log.exception( + 'Error while calling webhook "%s" for version %s.', + event.webhook.name, + version_pk, + ) + + if any(not result.is_complete for result in pending_results): + # Some scanners will answer later. Start a chain to wait for them: the + # one started when the version was created only waits for the events + # we aggregate. + log.info( + 'Waiting for the `on_version_scanned` results of version %s.', version_pk + ) + wait_for_scanner_results.apply_async( + kwargs={ + 'version_pk': version_pk, + 'event_ids': [WEBHOOK_ON_VERSION_SCANNED], + }, + countdown=settings.SCANNER_WEBHOOK_RETRY_DELAY, + ) + + +def scanner_webhook_events_to_wait_for(version): + """The events `version` is currently waiting on, i.e. which of the two + steps of `wait_for_scanner_results` it is in.""" + if is_waiting_on_scanner_webhook_events( + version=version, event_ids=WEBHOOK_EVENTS_AGGREGATED_BY_ON_VERSION_SCANNED + ): + return WEBHOOK_EVENTS_AGGREGATED_BY_ON_VERSION_SCANNED + return [WEBHOOK_ON_VERSION_SCANNED] + def is_waiting_on_scanner_webhook_events(*, version, event_ids): """Return whether `version` is still waiting on any active scanner webhook @@ -316,11 +469,15 @@ def wait_for_scanner_results(self, version_pk, event_ids=None): """Call the webhooks subscribed to `event_ids` that still owe us results again, retrying at a fixed interval until they answer, then record artificial results matching the SCANNER_RESULTS_MISSING rule to stop - waiting.""" + waiting. + + We wait in two steps because `on_version_scanned` cannot be sent before we + have the results it aggregates: a chain started for those events hands over + to a chain started for `on_version_scanned` once they are all in.""" # This is needed to handle tasks queued before we added the `event_ids` # param to this function's signature. if event_ids is None: - event_ids = WEBHOOK_EVENTS_BLOCKING_AUTO_APPROVAL + event_ids = WEBHOOK_EVENTS_AGGREGATED_BY_ON_VERSION_SCANNED if waffle.switch_is_active('disable-wait-for-scanner-results'): log.info( @@ -341,6 +498,9 @@ def wait_for_scanner_results(self, version_pk, event_ids=None): if not pending: log.info('All scanners have sent their results for version %s.', version_pk) + # The results `on_version_scanned` aggregates may be what we were + # waiting for. Does nothing once the event has been sent. + call_webhooks_on_version_scanned.delay(version_pk=version_pk) return give_up = self.request.retries >= WEBHOOK_MAX_RETRIES diff --git a/src/olympia/scanners/tests/test_commands.py b/src/olympia/scanners/tests/test_commands.py index d44d4405170b..8e05d0067a2c 100644 --- a/src/olympia/scanners/tests/test_commands.py +++ b/src/olympia/scanners/tests/test_commands.py @@ -15,8 +15,8 @@ from olympia.constants.scanners import ( WEBHOOK, WEBHOOK_DURING_VALIDATION, - WEBHOOK_EVENTS_BLOCKING_AUTO_APPROVAL, WEBHOOK_ON_SOURCE_CODE_UPLOADED, + WEBHOOK_ON_VERSION_SCANNED, ) from olympia.reviewers.models import AutoApprovalSummary from olympia.scanners.management.commands.retry_versions_waiting_on_scanners import ( @@ -210,8 +210,9 @@ def _run(self, *args): return stdout.getvalue() def test_min_age(self, delay_mock): - # 2h countdown, then a retry every 2h, 12 times. - assert MIN_AGE == timedelta(hours=26) + # 2h countdown, then a retry every 2h, 12 times, for each of the two + # steps we wait in. + assert MIN_AGE == timedelta(hours=52) def test_nothing_to_do(self, delay_mock): output = self._run('--force') @@ -224,8 +225,10 @@ def test_schedules_the_task(self, delay_mock): output = self._run('--force') + # No scanner owes us any of the results `on_version_scanned` + # aggregates, so the version is in the second step. delay_mock.assert_called_once_with( - version_pk=version.pk, event_ids=WEBHOOK_EVENTS_BLOCKING_AUTO_APPROVAL + version_pk=version.pk, event_ids=[WEBHOOK_ON_VERSION_SCANNED] ) assert 'Found 1 version(s) waiting on scanners (force=True).' in output assert f'version {version.pk}' in output diff --git a/src/olympia/scanners/tests/test_tasks.py b/src/olympia/scanners/tests/test_tasks.py index 2b5b67baf9d2..e7e065efad41 100644 --- a/src/olympia/scanners/tests/test_tasks.py +++ b/src/olympia/scanners/tests/test_tasks.py @@ -33,10 +33,12 @@ SCHEDULED, WEBHOOK, WEBHOOK_DURING_VALIDATION, - WEBHOOK_EVENTS_BLOCKING_AUTO_APPROVAL, + WEBHOOK_EVENTS_AGGREGATED_BY_ON_VERSION_SCANNED, WEBHOOK_MAX_RETRIES, WEBHOOK_ON_LISTING_CHANGED, + WEBHOOK_ON_SOURCE_CODE_UPLOADED, WEBHOOK_ON_VERSION_CREATED, + WEBHOOK_ON_VERSION_SCANNED, WEBHOOK_PUSH, YARA, ) @@ -61,7 +63,9 @@ call_webhooks, call_webhooks_during_validation, call_webhooks_on_listing_changed, + call_webhooks_on_version_scanned, mark_scanner_query_rule_as_completed_or_aborted, + maybe_call_webhooks_on_version_scanned, run_actions_for_scanner_result, run_narc_on_version, run_scanner_query_rule, @@ -3039,9 +3043,12 @@ def _create_result(self, **kwargs): kwargs.setdefault('version', self.version) return ScannerResult.objects.create(scanner=WEBHOOK, **kwargs) - def _run_task(self, retries=0): + def _run_task(self, retries=0, event_ids=None): return wait_for_scanner_results.apply( - args=(self.version.pk, WEBHOOK_EVENTS_BLOCKING_AUTO_APPROVAL), + args=( + self.version.pk, + event_ids or WEBHOOK_EVENTS_AGGREGATED_BY_ON_VERSION_SCANNED, + ), retries=retries, ) @@ -3256,3 +3263,398 @@ def test_records_missing_results_when_out_of_retries_and_no_payload( assert scanner_result.reload().results == { 'matchedRules': [SCANNER_RESULTS_MISSING_RULE_NAME] } + + +@mock.patch('olympia.scanners.tasks.wait_for_scanner_results.apply_async') +@mock.patch('olympia.scanners.tasks._call_webhook') +class TestCallWebhooksOnVersionScanned(UploadMixin, TestCase): + def setUp(self): + super().setUp() + + self.version = version_factory(addon=addon_factory()) + self.event = self._create_event( + WEBHOOK_ON_VERSION_SCANNED, name='aggregating-scanner' + ) + + def _create_event(self, event, *, name, created=None): + webhook = ScannerWebhook.objects.create( + name=name, + url='https://example.org/webhook', + api_key='some-api-key', + is_active=True, + ) + # The webhook has to predate the version, otherwise we don't wait on it. + webhook.update(modified=created or self.days_ago(1)) + return ScannerWebhookEvent.objects.create(event=event, webhook=webhook) + + def _create_result(self, event, **kwargs): + kwargs.setdefault('version', self.version) + return ScannerResult.objects.create( + scanner=WEBHOOK, webhook_event=event, **kwargs + ) + + def _run_task(self): + return call_webhooks_on_version_scanned(version_pk=self.version.pk) + + def _results_for(self, event): + return ScannerResult.objects.filter(version=self.version, webhook_event=event) + + def test_nothing_to_do_when_no_scanner_is_listening( + self, _call_webhook_mock, apply_async_mock + ): + self.event.update(is_active=False) + + self._run_task() + + _call_webhook_mock.assert_not_called() + assert not ScannerResult.objects.exists() + + def test_nothing_to_do_when_the_webhook_postdates_the_version( + self, _call_webhook_mock, apply_async_mock + ): + self.event.webhook.update(modified=datetime.now()) + + self._run_task() + + _call_webhook_mock.assert_not_called() + assert not ScannerResult.objects.exists() + + def test_nothing_to_do_while_an_aggregated_scanner_owes_us_results( + self, _call_webhook_mock, apply_async_mock + ): + self._create_result( + self._create_event(WEBHOOK_ON_VERSION_CREATED, name='some-scanner') + ) + + self._run_task() + + _call_webhook_mock.assert_not_called() + assert not self._results_for(self.event).exists() + + def test_calls_the_webhook_when_the_aggregated_scanners_are_done( + self, _call_webhook_mock, apply_async_mock + ): + _call_webhook_mock.return_value = {'matchedRules': []} + event = self._create_event(WEBHOOK_ON_VERSION_CREATED, name='some-scanner') + self._create_result(event, results={'matchedRules': ['SOME_RULE']}) + + self._run_task() + + scanner_result = self._results_for(self.event).get() + assert scanner_result.results == {'matchedRules': []} + assert _call_webhook_mock.call_count == 1 + assert _call_webhook_mock.call_args[1]['webhook'] == self.event.webhook + assert _call_webhook_mock.call_args[1]['request_id'] is not None + payload = _call_webhook_mock.call_args[1]['payload'] + assert payload.keys() == { + 'addon', + 'version', + 'scanner_results', + 'event', + 'scanner_result_url', + } + assert payload['addon']['id'] == self.version.addon.pk + assert payload['version']['id'] == self.version.pk + assert payload['scanner_results'] == { + 'during_validation': {}, + 'on_version_created': {'some-scanner': {'matchedRules': ['SOME_RULE']}}, + } + assert payload['event'] == 'on_version_scanned' + assert payload['scanner_result_url'] == ( + f'http://testserver/api/v5/scanner/results/{scanner_result.pk}/' + ) + + def test_calls_the_webhook_when_no_scanner_is_aggregated( + self, _call_webhook_mock, apply_async_mock + ): + _call_webhook_mock.return_value = {'matchedRules': []} + + self._run_task() + + assert self._results_for(self.event).exists() + assert _call_webhook_mock.call_args[1]['payload']['scanner_results'] == { + 'during_validation': {}, + 'on_version_created': {}, + } + + def test_aggregates_every_event_and_scanner( + self, _call_webhook_mock, apply_async_mock + ): + _call_webhook_mock.return_value = {} + during_validation = self._create_event( + WEBHOOK_DURING_VALIDATION, name='some-scanner' + ) + # The same scanner listening to both aggregated events. + on_version_created = ScannerWebhookEvent.objects.create( + event=WEBHOOK_ON_VERSION_CREATED, webhook=during_validation.webhook + ) + other = self._create_event(WEBHOOK_ON_VERSION_CREATED, name='other-scanner') + self._create_result( + during_validation, + results={'matchedRules': ['SOME_RULE']}, + upload=self.get_upload('webextension.xpi'), + ) + self._create_result(on_version_created, results={'matchedRules': []}) + # This scanner skipped the event. + self._create_result(other, results=None) + + self._run_task() + + assert _call_webhook_mock.call_args[1]['payload']['scanner_results'] == { + 'during_validation': {'some-scanner': {'matchedRules': ['SOME_RULE']}}, + 'on_version_created': { + 'some-scanner': {'matchedRules': []}, + 'other-scanner': None, + }, + } + + def test_aggregates_the_results_of_a_deactivated_scanner( + self, _call_webhook_mock, apply_async_mock + ): + _call_webhook_mock.return_value = {} + event = self._create_event(WEBHOOK_ON_VERSION_CREATED, name='some-scanner') + self._create_result(event, results={'matchedRules': []}) + # We already have its results, so we keep reporting them. + event.webhook.update(is_active=False) + + self._run_task() + + assert _call_webhook_mock.call_args[1]['payload']['scanner_results'] == { + 'during_validation': {}, + 'on_version_created': {'some-scanner': {'matchedRules': []}}, + } + + def test_aggregates_the_last_result_of_a_scanner( + self, _call_webhook_mock, apply_async_mock + ): + _call_webhook_mock.return_value = {} + event = self._create_event(WEBHOOK_ON_VERSION_CREATED, name='some-scanner') + self._create_result(event, results={'matchedRules': ['FIRST']}) + self._create_result(event, results={'matchedRules': ['LAST']}) + + self._run_task() + + assert _call_webhook_mock.call_args[1]['payload']['scanner_results'][ + 'on_version_created' + ] == {'some-scanner': {'matchedRules': ['LAST']}} + + def test_does_not_aggregate_incomplete_results( + self, _call_webhook_mock, apply_async_mock + ): + _call_webhook_mock.return_value = {} + event = self._create_event(WEBHOOK_ON_VERSION_CREATED, name='some-scanner') + self._create_result(event, results={'matchedRules': []}) + # Another event for a scanner we are not waiting on (it was added after + # the version was created), with no results yet. + self._create_result( + self._create_event( + WEBHOOK_ON_VERSION_CREATED, + name='other-scanner', + created=datetime.now(), + ) + ) + + self._run_task() + + assert _call_webhook_mock.call_args[1]['payload']['scanner_results'][ + 'on_version_created' + ] == {'some-scanner': {'matchedRules': []}} + + def test_does_not_call_the_webhook_twice( + self, _call_webhook_mock, apply_async_mock + ): + _call_webhook_mock.return_value = {} + + self._run_task() + self._run_task() + + assert _call_webhook_mock.call_count == 1 + assert self._results_for(self.event).count() == 1 + + def test_calls_every_subscribed_webhook(self, _call_webhook_mock, apply_async_mock): + _call_webhook_mock.return_value = {} + other_event = self._create_event( + WEBHOOK_ON_VERSION_SCANNED, name='other-aggregating-scanner' + ) + + self._run_task() + + assert _call_webhook_mock.call_count == 2 + assert [call[1]['webhook'] for call in _call_webhook_mock.call_args_list] == [ + self.event.webhook, + other_event.webhook, + ] + + def test_does_not_stop_on_a_webhook_error( + self, _call_webhook_mock, apply_async_mock + ): + _call_webhook_mock.side_effect = [ValueError('scanner is down'), {}] + other_event = self._create_event( + WEBHOOK_ON_VERSION_SCANNED, name='other-aggregating-scanner' + ) + + self._run_task() + + assert _call_webhook_mock.call_count == 2 + assert self._results_for(self.event).get().results == [] + assert self._results_for(other_event).get().results == {} + + def test_waits_for_the_results_it_just_asked_for( + self, _call_webhook_mock, apply_async_mock + ): + # A `202` with no results: the scanner will send them later. + _call_webhook_mock.return_value = {} + self._create_result( + self._create_event(WEBHOOK_ON_VERSION_CREATED, name='some-scanner'), + results={'matchedRules': []}, + ) + + self._run_task() + + assert not self._results_for(self.event).get().is_complete + # Nothing was waiting for this result before we created it: the chain + # for the events we aggregate only ever watched those. + apply_async_mock.assert_called_once_with( + kwargs={ + 'version_pk': self.version.pk, + 'event_ids': [WEBHOOK_ON_VERSION_SCANNED], + }, + countdown=settings.SCANNER_WEBHOOK_RETRY_DELAY, + ) + + def test_does_not_wait_when_the_scanner_answered_synchronously( + self, _call_webhook_mock, apply_async_mock + ): + _call_webhook_mock.return_value = {'matchedRules': []} + self._create_result( + self._create_event(WEBHOOK_ON_VERSION_CREATED, name='some-scanner'), + results={'matchedRules': []}, + ) + + self._run_task() + + assert self._results_for(self.event).get().is_complete + apply_async_mock.assert_not_called() + + def test_does_not_wait_when_there_is_nothing_to_send( + self, _call_webhook_mock, apply_async_mock + ): + self.event.update(is_active=False) + + self._run_task() + + _call_webhook_mock.assert_not_called() + apply_async_mock.assert_not_called() + + +@mock.patch('olympia.scanners.tasks.call_webhooks_on_version_scanned.delay') +class TestMaybeCallWebhooksOnVersionScanned(TestCase): + def setUp(self): + super().setUp() + + self.version = version_factory(addon=addon_factory()) + webhook = ScannerWebhook.objects.create( + name='some-scanner', + url='https://example.org/webhook', + api_key='some-api-key', + is_active=True, + ) + webhook.update(modified=self.days_ago(1)) + self.event = ScannerWebhookEvent.objects.create( + event=WEBHOOK_ON_VERSION_CREATED, webhook=webhook + ) + + def _create_result(self, **kwargs): + kwargs.setdefault('version', self.version) + kwargs.setdefault('webhook_event', self.event) + return ScannerResult.objects.create(scanner=WEBHOOK, **kwargs) + + def test_schedules_the_task(self, delay_mock): + maybe_call_webhooks_on_version_scanned( + self._create_result(results={'matchedRules': []}) + ) + + delay_mock.assert_called_once_with(version_pk=self.version.pk) + + def test_schedules_the_task_when_the_event_was_skipped(self, delay_mock): + maybe_call_webhooks_on_version_scanned(self._create_result(results=None)) + + delay_mock.assert_called_once_with(version_pk=self.version.pk) + + def test_does_nothing_when_the_result_is_incomplete(self, delay_mock): + maybe_call_webhooks_on_version_scanned(self._create_result()) + + delay_mock.assert_not_called() + + def test_does_nothing_without_a_version(self, delay_mock): + maybe_call_webhooks_on_version_scanned( + self._create_result(version=None, results={'matchedRules': []}) + ) + + delay_mock.assert_not_called() + + def test_does_nothing_for_an_event_that_is_not_aggregated(self, delay_mock): + self.event.update(event=WEBHOOK_ON_SOURCE_CODE_UPLOADED) + + maybe_call_webhooks_on_version_scanned( + self._create_result(results={'matchedRules': []}) + ) + + delay_mock.assert_not_called() + + @mock.patch('olympia.scanners.tasks._call_webhook') + def test_scheduled_when_a_webhook_answers_synchronously( + self, _call_webhook_mock, delay_mock + ): + _call_webhook_mock.return_value = {'matchedRules': []} + + call_webhooks( + event_id=WEBHOOK_ON_VERSION_CREATED, + payload={}, + version=self.version, + ) + + delay_mock.assert_called_with(version_pk=self.version.pk) + + @mock.patch('olympia.scanners.tasks._call_webhook') + def test_not_scheduled_when_a_webhook_answers_asynchronously( + self, _call_webhook_mock, delay_mock + ): + # A `202` with no results: the scanner will send them later. + _call_webhook_mock.return_value = {} + + call_webhooks( + event_id=WEBHOOK_ON_VERSION_CREATED, + payload={}, + version=self.version, + ) + + delay_mock.assert_not_called() + + @mock.patch('olympia.scanners.tasks._call_webhook') + def test_scheduled_by_wait_for_scanner_results( + self, _call_webhook_mock, delay_mock + ): + _call_webhook_mock.return_value = {} + # Nothing left to wait for, which is what unlocks `on_version_scanned`. + self._create_result(results={'matchedRules': []}) + + wait_for_scanner_results.apply( + args=(self.version.pk, WEBHOOK_EVENTS_AGGREGATED_BY_ON_VERSION_SCANNED) + ) + + delay_mock.assert_called_with(version_pk=self.version.pk) + + @mock.patch('olympia.scanners.tasks._call_webhook') + def test_not_scheduled_while_wait_for_scanner_results_is_still_waiting( + self, _call_webhook_mock, delay_mock + ): + _call_webhook_mock.return_value = {} + self._create_result() + + with pytest.raises(Retry): + wait_for_scanner_results.apply( + args=(self.version.pk, WEBHOOK_EVENTS_AGGREGATED_BY_ON_VERSION_SCANNED) + ) + + delay_mock.assert_not_called() diff --git a/src/olympia/versions/tasks.py b/src/olympia/versions/tasks.py index 811738af65cb..1392d54d1049 100644 --- a/src/olympia/versions/tasks.py +++ b/src/olympia/versions/tasks.py @@ -19,7 +19,7 @@ from olympia.amo.decorators import use_primary_db from olympia.amo.utils import SafeStorage, extract_colors_from_image, pngcrush_image from olympia.constants.scanners import ( - WEBHOOK_EVENTS_BLOCKING_AUTO_APPROVAL, + WEBHOOK_EVENTS_AGGREGATED_BY_ON_VERSION_SCANNED, WEBHOOK_ON_SOURCE_CODE_UPLOADED, WEBHOOK_ON_VERSION_CREATED, ) @@ -32,6 +32,7 @@ from olympia.scanners.tasks import ( build_webhook_payload, call_webhooks, + call_webhooks_on_version_scanned, wait_for_scanner_results, ) from olympia.users.models import UserProfile @@ -455,12 +456,18 @@ def call_webhooks_on_version_created(version_pk): except Exception: log.exception('Error while calling webhooks for Version %s', version_pk) - # Even when a webhook call failed, we still want to wait for the results of - # every scanner blocking the auto-approval of this version. + # The scanners called above may all have answered synchronously, or none + # may have been called at all, in which case nothing else would send + # `on_version_scanned` before the retries below. + call_webhooks_on_version_scanned.delay(version_pk=version_pk) + + # Even when a webhook call failed, we still want to wait for the results + # `on_version_scanned` aggregates. Waiting for `on_version_scanned` itself + # is a second step, started once we have them. wait_for_scanner_results.apply_async( kwargs={ 'version_pk': version_pk, - 'event_ids': WEBHOOK_EVENTS_BLOCKING_AUTO_APPROVAL, + 'event_ids': WEBHOOK_EVENTS_AGGREGATED_BY_ON_VERSION_SCANNED, }, countdown=settings.SCANNER_WEBHOOK_RETRY_DELAY, ) diff --git a/src/olympia/versions/tests/test_tasks.py b/src/olympia/versions/tests/test_tasks.py index eee9c3801072..cda2b0ee26fd 100644 --- a/src/olympia/versions/tests/test_tasks.py +++ b/src/olympia/versions/tests/test_tasks.py @@ -26,7 +26,7 @@ from olympia.blocklist.models import Block, BlockType, BlockVersion from olympia.constants.blocklist import BlockReason from olympia.constants.scanners import ( - WEBHOOK_EVENTS_BLOCKING_AUTO_APPROVAL, + WEBHOOK_EVENTS_AGGREGATED_BY_ON_VERSION_SCANNED, WEBHOOK_ON_SOURCE_CODE_UPLOADED, WEBHOOK_ON_VERSION_CREATED, ) @@ -1071,7 +1071,7 @@ def test_schedules_wait_for_scanner_results(self, _, apply_async_mock): apply_async_mock.assert_called_with( kwargs={ 'version_pk': version.pk, - 'event_ids': WEBHOOK_EVENTS_BLOCKING_AUTO_APPROVAL, + 'event_ids': WEBHOOK_EVENTS_AGGREGATED_BY_ON_VERSION_SCANNED, }, countdown=settings.SCANNER_WEBHOOK_RETRY_DELAY, ) @@ -1088,6 +1088,16 @@ def test_schedules_wait_for_scanner_results_even_when_a_call_failed( assert apply_async_mock.called + @mock.patch('olympia.versions.tasks.call_webhooks_on_version_scanned.delay') + @mock.patch('olympia.versions.tasks.wait_for_scanner_results.apply_async') + @mock.patch('olympia.versions.tasks.call_webhooks') + def test_calls_webhooks_on_version_scanned(self, _, __, delay_mock): + version = version_factory(addon=addon_factory()) + + call_webhooks_on_version_created(version.pk) + + delay_mock.assert_called_once_with(version_pk=version.pk) + @mock.patch('olympia.versions.tasks.wait_for_scanner_results.apply_async') @mock.patch('olympia.versions.tasks.call_webhooks') def test_skip_webhooks_on_langpack_created(