fix(console): classify Worker annotations without hiding drift

python/tests/test_console_config_compare.py

@@ -176,7 +176,7 @@ def test_unsupported_fields_emit_only_sanitized_names_and_remain_blocked(self):
         cases = [('settings', 'unsupported_live_setting'), ('config', 'unsupported_config_key'),
                  ('subdomain', 'unsupported_subdomain_setting'), ('runtime', 'unmapped_runtime_setting')]
         for scope, reason in cases:
-            for name in ('annotations', 'A_0.key-name', 'x' * 64, '', 'x' * 65,
+            for name in ('future_setting', 'A_0.key-name', 'x' * 64, '', 'x' * 65,
                          'bad/name', 'bad=PRIVATE_TOKEN', 'bad\nPRIVATE_TOKEN', '非法字段'):
                 with self.subTest(scope=scope, name=name):
                     self.config, self.live = fixture()
@@ -186,7 +186,7 @@ def test_unsupported_fields_emit_only_sanitized_names_and_remain_blocked(self):
                     rows = self.rows()
                     row = next(row for row in rows if row.get('reason') == reason)
                     self.assertEqual(row['status'], 'unreadable')
-                    valid = name in ('annotations', 'A_0.key-name', 'x' * 64)
+                    valid = name in ('future_setting', 'A_0.key-name', 'x' * 64)
                     self.assertEqual(row['field'], name if valid else '<unsupported-field>')
                     output = json.dumps(rows, ensure_ascii=False)
                     for private in ('PRIVATE_TOKEN', 'PRIVATE_SECRET', 'nested_secret', 'account_id',
@@ -195,6 +195,122 @@ def test_unsupported_fields_emit_only_sanitized_names_and_remain_blocked(self):
                     if not valid and name:
                         self.assertNotIn(name, output)
 
+    def test_known_annotations_are_recorded_as_redacted_deployment_metadata(self):
+        for annotations in ({}, {'workers/message': 'PRIVATE_MESSAGE'},
+                            {'workers/tag': 'PRIVATE_TAG'}, {'workers/triggered_by': 'PRIVATE_TRIGGER'},
+                            {'workers/message': 'PRIVATE_MESSAGE', 'workers/tag': 'PRIVATE_TAG',
+                             'workers/triggered_by': 'PRIVATE_TRIGGER'}):
+            with self.subTest(keys=sorted(annotations)):
+                self.live['settings']['annotations'] = annotations
+                rows = self.rows()
+                self.assertTrue(all(row['status'] == 'match' for row in rows), rows)
+                recorded = [row for row in rows if row['item'].startswith('settings.annotations')]
+                self.assertTrue(recorded)
+                self.assertTrue(all(row['classification'] == 'deployment_metadata' for row in recorded))
+                for key in annotations:
+                    row = next(row for row in recorded if row['item'] == 'settings.annotations.' + key)
+                    self.assertEqual(row['live'], '<redacted>')
+                    self.assertEqual(row['rendered'], 'absent')
+                self.assertNotIn('PRIVATE_', json.dumps(rows))
+
+    def test_unknown_annotation_keys_including_alias_block_without_echoing_keys(self):
+        for key in ('workers/alias', 'workers/future', 'PRIVATE_KEY', 'bad\nPRIVATE_KEY', ''):
+            with self.subTest(index=len(key)):
+                self.live['settings']['annotations'] = {
+                    'workers/message': 'PRIVATE_MESSAGE', key: {'nested': 'PRIVATE_SECRET'}}
+                rows = self.rows()
+                blocked = [row for row in rows if row.get('reason') == 'unsupported_annotation_key']
+                self.assertEqual(len(blocked), 1)
+                self.assertEqual(blocked[0]['status'], 'unreadable')
+                output = json.dumps(rows)
+                self.assertNotIn('PRIVATE_', output)
+                self.assertNotIn('workers/alias', output)
+                self.assertNotIn('workers/future', output)
+
+    def test_malformed_annotations_and_non_string_known_values_block(self):
+        for value in (None, [], ['PRIVATE_SECRET'], 'PRIVATE_SECRET', False, 1):
+            with self.subTest(shape=type(value).__name__):
+                self.live['settings']['annotations'] = value
+                rows = self.rows()
+                self.assertTrue(any(row.get('reason') == 'unsupported_annotations_shape' for row in rows))
+                self.assertNotIn('PRIVATE_', json.dumps(rows))
+        for key in ('workers/message', 'workers/tag', 'workers/triggered_by'):
+            for value in (None, True, 1, [], {'PRIVATE_KEY': 'PRIVATE_SECRET'}):
+                with self.subTest(key=key, shape=type(value).__name__):
+                    self.live['settings']['annotations'] = {key: value}
+                    rows = self.rows()
+                    self.assertTrue(any(row.get('reason') == 'unsupported_annotation_value' for row in rows))
+                    self.assertNotIn('PRIVATE_', json.dumps(rows))
+
+    def test_annotation_limits_use_utf8_bytes_and_reject_invalid_unicode(self):
+        for key, limit in (('workers/message', 1000), ('workers/tag', 100)):
+            for value, valid in (('', True), ('x' * limit, True), ('x' * (limit + 1), False),
+                                 ('\u00e9' * (limit // 2), True), ('\u00e9' * (limit // 2 + 1), False),
+                                 ('\ud800', False)):
+                with self.subTest(key=key, length=len(value), valid=valid):
+                    self.live['settings']['annotations'] = {key: value}
+                    row = next(row for row in self.rows() if row['item'] == 'settings.annotations.' + key)
+                    self.assertEqual(row['status'], 'match' if valid else 'unreadable')
+
+    def test_version_tag_blocks_if_either_inventory_exposes_version_metadata(self):
+        for scope in ('settings', 'version_bindings'):
+            with self.subTest(scope=scope):
+                self.config, self.live = fixture()
+                self.live['settings']['annotations'] = {'workers/tag': 'PRIVATE_TAG'}
+                bindings = self.live['settings']['bindings'] if scope == 'settings' else self.live[scope]
+                bindings.append({'name': 'PRIVATE_BINDING', 'type': 'version_metadata'})
+                rows = self.rows()
+                row = next(row for row in rows if row['item'] == 'settings.annotations.workers/tag')
+                self.assertEqual(row['status'], 'unreadable')
+                self.assertEqual(row['reason'], 'annotation_runtime_dependency')
+                self.assertNotIn('PRIVATE_', json.dumps(rows))
+
+    def test_version_tag_requires_readable_binding_inventories(self):
+        for scope in ('settings', 'version_bindings'):
+            for value in (None, [None]):
+                with self.subTest(scope=scope, shape=type(value).__name__):
+                    self.config, self.live = fixture()
+                    self.live['settings']['annotations'] = {'workers/tag': 'PRIVATE_TAG'}
+                    if scope == 'settings':
+                        self.live[scope]['bindings'] = value
+                    else:
+                        self.live[scope] = value
+                    row = next(row for row in self.rows() if row['item'] == 'settings.annotations.workers/tag')
+                    self.assertEqual(row['status'], 'unreadable')
+
+    def test_annotations_do_not_allow_local_or_runtime_configuration(self):
+        for scope in ('config', 'runtime'):
+            with self.subTest(scope=scope):
+                self.config, self.live = fixture()
+                target = self.config if scope == 'config' else self.live[scope]
+                target['annotations'] = {'workers/message': 'PRIVATE_MESSAGE'}
+                self.assertBlocked()
+                self.assertNotIn('PRIVATE_', json.dumps(self.rows()))
+
+    def test_annotation_change_during_observation_still_blocks(self):
+        self.live['settings']['annotations'] = {'workers/message': 'PRIVATE_BEFORE'}
+        after = copy.deepcopy(self.live)
+        after['settings']['annotations']['workers/message'] = 'PRIVATE_AFTER'
+        with patch.object(compare, 'snapshot', side_effect=[self.live, after]):
+            rows = compare.run(self.config, object())
+        self.assertEqual(rows[-1], {'item': 'observation.stable', 'status': 'mismatch'})
+        self.assertNotIn('PRIVATE_', json.dumps(rows))
+
+    def test_annotation_values_never_reach_main_output_and_unknown_still_exits_nonzero(self):
+        for unknown in (False, True):
+            with self.subTest(unknown=unknown):
+                self.live['settings']['annotations'] = {'workers/message': 'PRIVATE_SECRET'}
+                if unknown:
+                    self.live['settings']['annotations']['PRIVATE_KEY'] = 'PRIVATE_SECRET'
+                with patch.object(Path, 'read_text', return_value=''), \
+                        patch.object(compare.tomllib, 'loads', return_value=self.config), \
+                        patch.dict(os.environ, {'CLOUDFLARE_API_TOKEN': 'PRIVATE_TOKEN'}), \
+                        patch.object(compare, 'snapshot', return_value=self.live), \
+                        contextlib.redirect_stdout(io.StringIO()) as out:
+                    self.assertEqual(compare.main(), 1 if unknown else 0)
+                self.assertEqual(json.loads(out.getvalue())['all_match'], not unknown)
+                self.assertNotIn('PRIVATE_', out.getvalue())
+
     def test_each_binding_value_change_or_removal_blocks(self):
         for index, field in [(0, 'text'), (1, 'namespace_id'), (2, 'class_name'), (3, 'name')]:
             with self.subTest(index=index):

scripts/console_config_compare.py

@@ -36,6 +36,11 @@
     'placement', 'logpush', 'observability', 'tail_consumers', 'tags',
     'cache_options', 'streaming_tail_consumers',
 }
+# Cloudflare script/version annotations, not Worker tags or runtime settings.
+# Message/tag limits are UTF-8 bytes; triggered_by is server-provided provenance.
+ANNOTATION_BYTE_LIMITS = {
+    'workers/message': 1000, 'workers/tag': 100, 'workers/triggered_by': None,
+}
 
 
 class Unreadable(Exception):
@@ -185,12 +190,13 @@ def snapshot(reader, name):
 def compare(config, live):
     rows = []
 
-    def row(item, expected=MISSING, actual=MISSING, reason=None, *, field=MISSING):
-        status = 'unreadable' if reason else ('match' if expected == actual else 'mismatch')
+    def row(item, expected=MISSING, actual=MISSING, reason=None, *, field=MISSING, metadata=False):
+        status = 'unreadable' if reason else ('match' if metadata or expected == actual else 'mismatch')
         rows.append({'item': item, 'status': status,
                      'rendered': 'absent' if expected is MISSING else '<redacted>',
                      'live': 'absent' if actual is MISSING else '<redacted>',
                      **({'reason': reason} if reason else {}),
+                     **({'classification': 'deployment_metadata'} if metadata else {}),
                      **({'field': field if isinstance(field, str) and re.fullmatch(r'[A-Za-z0-9_.-]{1,64}', field)
                          else '<unsupported-field>'} if field is not MISSING else {})})
 
@@ -317,7 +323,44 @@ def row(item, expected=MISSING, actual=MISSING, reason=None, *, field=MISSING):
             row('settings.' + key, expected, actual, 'undeclared_effective_setting')
         else:
             row('settings.' + key, expected, actual)
-    for index, key in enumerate(sorted(set(settings) - SETTING_KEYS - {'bindings'})):
+    annotations = settings.get('annotations', MISSING)
+    if annotations is MISSING or annotations == {}:
+        row('settings.annotations', actual=annotations, metadata=True)
+    elif not isinstance(annotations, dict):
+        row('settings.annotations', actual=annotations, reason='unsupported_annotations_shape')
+    else:
+        # Version tags can be consumed by Worker code via version_metadata.
+        # Do not classify them as harmless without both binding inventories.
+        inventories = (settings.get('bindings'), live.get('version_bindings'))
+        metadata_unexposed = all(
+            isinstance(inventory, list) and all(
+                isinstance(binding, dict) and isinstance(binding.get('name'), str)
+                and isinstance(binding.get('type'), str) and binding['type'] != 'version_metadata'
+                for binding in inventory)
+            for inventory in inventories)
+        for index, key in enumerate(sorted(annotations)):
+            actual = annotations[key]
+            if key not in ANNOTATION_BYTE_LIMITS:
+                # Unknown keys can themselves contain private data. Never echo
+                # them, including workers/alias, which can affect version URLs.
+                row(f'settings.annotations.unsupported[{index}]', actual=actual,
+                    reason='unsupported_annotation_key')
+                continue
+            limit = ANNOTATION_BYTE_LIMITS[key]
+            valid = isinstance(actual, str)
+            if valid:
+                try:
+                    size = len(actual.encode('utf-8'))
+                    valid = limit is None or size <= limit
+                except UnicodeError:
+                    valid = False
+            reason = None
+            if not valid:
+                reason = 'unsupported_annotation_value'
+            elif key == 'workers/tag' and not metadata_unexposed:
+                reason = 'annotation_runtime_dependency'
+            row('settings.annotations.' + key, actual=actual, reason=reason, metadata=reason is None)
+    for index, key in enumerate(sorted(set(settings) - SETTING_KEYS - {'bindings', 'annotations'})):
         row(f'settings.unsupported[{index}]', reason='unsupported_live_setting', field=key)
     migrations = config.get('migrations', [])
     if len(migrations) != 1 or set(migrations[0]) != {'tag', 'new_sqlite_classes'}:

web/strategy-switch-console/README.md

@@ -346,6 +346,30 @@ name: ASCII letters, digits, `_`, `.`, or `-`, with 1–64 characters. Other nam
 become `<unsupported-field>`; values and nested content are never emitted.
 This diagnostic does not make the field supported.
 
+The settings response's optional `annotations` object describes the Worker
+version, separately from Worker `tags`. Only string-valued `workers/message`
+(up to 1000 UTF-8 bytes), `workers/tag` (up to 100 bytes), and
+`workers/triggered_by` are recognized. They describe the version message,
+identifier and creation operation; `triggered_by` is set by the server.
+Recognized entries produce `match` rows with
+`classification=deployment_metadata`, an absent rendered value and a redacted
+live value. They are recorded without treating deployment metadata as config
+drift. Omission and an empty object are also accepted; null, malformed objects,
+non-string/invalid values and unknown keys remain unreadable. Unknown annotation
+keys and all annotation values are redacted. Local `annotations` declarations
+and annotations inside runtime settings are still unsupported.
+
+A version tag can be exposed to Worker code through a `version_metadata`
+binding. A tag is therefore accepted as deployment metadata only when both the
+settings and active-version binding inventories are readable and neither has
+that binding; otherwise `annotation_runtime_dependency` blocks comparison.
+Other keys, including `workers/alias` (which can affect version URLs), are never
+implicitly allowed. The existing full-snapshot stability check still blocks
+changes between observations, including metadata changes. See the
+[settings API](https://developers.cloudflare.com/api/typescript/resources/workers/subresources/scripts/subresources/script_and_version_settings/),
+[upload annotations](https://developers.cloudflare.com/workers/configuration/multipart-upload-metadata/)
+and [runtime version metadata](https://developers.cloudflare.com/workers/runtime-apis/bindings/version-metadata/).
+
 An omitted or empty rendered `compatibility_flags` list matches an omitted,
 null or empty list in the settings and active runtime responses. Nonempty flags
 still require an exact match; invalid shapes and missing required dates block