Coverage for sources/emcdproj/website/inventory.py: 86%
113 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-14 22:01 +0000
1# vim: set filetype=python fileencoding=utf-8:
2# -*- coding: utf-8 -*-
4#============================================================================#
5# #
6# Licensed under the Apache License, Version 2.0 (the "License"); #
7# you may not use this file except in compliance with the License. #
8# You may obtain a copy of the License at #
9# #
10# http://www.apache.org/licenses/LICENSE-2.0 #
11# #
12# Unless required by applicable law or agreed to in writing, software #
13# distributed under the License is distributed on an "AS IS" BASIS, #
14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. #
15# See the License for the specific language governing permissions and #
16# limitations under the License. #
17# #
18#============================================================================#
21''' Publication validation: inventory, history, and anchors. '''
24from . import __
25from . import locations as _locations
26from . import tracks as _tracks
28_MUTABLE_TOP_LEVEL: tuple[ str, ... ] = (
29 'index.html', 'versions.json', 'provenance.json',
30 'coverage.svg', '.nojekyll' )
33_VOLATILE_VERSION_FILES: tuple[ str, ... ] = (
34 'coverage.xml', '.buildinfo' )
37def snapshot_archive( archive: __.Path ) -> dict[ str, str ]:
38 ''' Snapshots retained-history digests from the trusted archive.
40 Reads the pre-update archive members without extracting, so
41 restore loss itself is visible to later comparison. Returns an
42 empty manifest when no archive exists yet.
43 '''
44 import hashlib
45 import tarfile
46 if not archive.is_file( ): return { }
47 manifest: dict[ str, str ] = { }
48 try:
49 with tarfile.open( archive, mode = 'r:*' ) as bundle:
50 for member in bundle.getmembers( ):
51 if not member.isfile( ): continue
52 name = member.name.replace( '\\', '/' )
53 while name.startswith( './' ): name = name[ 2 : ]
54 if not name or name == '.': continue 54 ↛ 50line 54 didn't jump to line 50 because the continue on line 54 wasn't executed
55 handle = bundle.extractfile( member )
56 if handle is None: continue 56 ↛ 50line 56 didn't jump to line 50 because the continue on line 56 wasn't executed
57 manifest[ name ] = hashlib.sha256(
58 handle.read( ) ).hexdigest( )
59 except ( tarfile.TarError, OSError, EOFError ) as exception:
60 raise __.exceptions.PublicationSourceInvalidity(
61 str( archive ), 'unreadable archive' ) from exception
62 return manifest
65def reject_alias_collisions(
66 index_data: dict[ __.typx.Any, __.typx.Any ],
67 version: str,
68) -> None:
69 ''' Rejects aliases colliding with retained releases.
71 An alias pointing at a different retained version would overwrite
72 history that the retained check then skips. Only deliberate
73 identity (alias equal to its selected version) is allowed.
74 '''
75 mapping = index_data.get( 'track_versions', { } )
76 retained = {
77 label for label in index_data.get( 'versions', { } )
78 if label != version }
79 for alias, selected in mapping.items( ):
80 if alias != selected and alias in retained:
81 raise __.exceptions.PublicationSourceInvalidity(
82 alias, 'alias collides with retained release' )
83 if version in mapping and mapping.get( version ) != version: 83 ↛ 84line 83 didn't jump to line 84 because the condition on line 83 was never true
84 raise __.exceptions.PublicationSourceInvalidity(
85 version, 'version collides with alias' )
88def verify_retained_history(
89 locations: _locations.Locations,
90 snapshot: dict[ str, str ],
91 version: str,
92 tracks: _tracks.TrackConfiguration,
93) -> None:
94 ''' Verifies retained history against the pre-build snapshot.
96 Every snapshotted file outside the published version, configured
97 aliases, and mutable top-level files must exist with identical
98 content. Republished versions must reproduce retained files
99 byte-identically apart from volatile build outputs; changed
100 same-version content is refused while idempotent reruns pass.
101 Legitimate new version files and mutable outputs are excluded by
102 explicit class rules, never by regeneration.
103 '''
104 import hashlib
105 aliases = {
106 track.get( 'alias', name ) for name, track in tracks.items( ) }
107 for relative, digest in snapshot.items( ):
108 head = relative.split( '/', 1 )[ 0 ]
109 if head in aliases and head != version: continue
110 if '/' not in relative and relative in _MUTABLE_TOP_LEVEL: continue
111 if head == version:
112 base = relative.rsplit( '/', 1 )[ -1 ]
113 if base in _VOLATILE_VERSION_FILES: continue
114 candidate = locations.website / relative
115 if not candidate.is_file( ): 115 ↛ 116line 115 didn't jump to line 116 because the condition on line 115 was never true
116 raise __.exceptions.PublicationSourceInvalidity( relative,
117 'history lost' )
118 actual = hashlib.sha256( candidate.read_bytes( ) ).hexdigest( )
119 if actual != digest:
120 raise __.exceptions.PublicationSourceInvalidity( relative,
121 'history drift' )
124def _validate_inventory_path( root: __.Path, entry: str ) -> __.Path:
125 ''' Resolves inventory entry path under website root. '''
126 if entry.startswith( '/' ) or entry.startswith( '\\' ): 126 ↛ 127line 126 didn't jump to line 127 because the condition on line 126 was never true
127 raise __.exceptions.PublicationSourceInvalidity( entry,
128 'absolute path rejected' )
129 parts = entry.replace( '\\', '/' ).split( '/' )
130 if '..' in parts or not entry:
131 raise __.exceptions.PublicationSourceInvalidity( entry,
132 'traversal rejected' )
133 return root / entry.replace( '/', __.os.sep )
136def validate_legacy_inventory(
137 locations: _locations.Locations, inventory: __.Path | str
138) -> None:
139 ''' Validates legacy-link inventory against built tree.
141 Every entry must resolve; frozen entries require matching content
142 digests; listed fragments must occur as element IDs. Malformed
143 staged input aborts promotion without changing prior output.
144 '''
145 path = __.Path( inventory )
146 try: content = path.read_bytes( )
147 except OSError as exception:
148 raise __.exceptions.FileAwol( path ) from exception
149 try: data = _tracks.toml_loads( content )
150 except Exception as exception:
151 raise __.exceptions.PublicationSourceInvalidity(
152 str( path ), 'unparseable inventory' ) from exception
153 try: items = list( data.get( 'entries', [ ] ) )
154 except TypeError as exception:
155 raise __.exceptions.PublicationSourceInvalidity(
156 str( path ), 'malformed inventory' ) from exception
157 if not items: 157 ↛ 158line 157 didn't jump to line 158 because the condition on line 157 was never true
158 raise __.exceptions.PublicationSourceInvalidity(
159 str( path ), 'empty inventory' )
160 for raw_entry in items:
161 _validate_inventory_entry( locations, raw_entry )
164def _validate_inventory_entry(
165 locations: _locations.Locations, raw_entry: __.typx.Any
166) -> None:
167 ''' Validates one legacy inventory entry against built tree.
169 Consumes dynamic TOML values validated at runtime via duck typing;
170 malformed staged input aborts promotion.
171 '''
172 get = getattr( raw_entry, 'get', None )
173 if get is None: 173 ↛ 174line 173 didn't jump to line 174 because the condition on line 173 was never true
174 raise __.exceptions.PublicationSourceInvalidity( 'entry',
175 'malformed entry' )
176 relative = get( 'path', '' )
177 semantics = get( 'semantics', 'frozen' )
178 if not isinstance( relative, str ) or not isinstance( semantics, str ): 178 ↛ 179line 178 didn't jump to line 179 because the condition on line 178 was never true
179 raise __.exceptions.PublicationSourceInvalidity( 'entry',
180 'malformed entry' )
181 if semantics not in ( 'frozen', 'follow' ): 181 ↛ 182line 181 didn't jump to line 182 because the condition on line 181 was never true
182 raise __.exceptions.PublicationSourceInvalidity( relative,
183 'unknown semantics' )
184 target = _validate_inventory_path( locations.website, relative )
185 if not target.is_file( ):
186 raise __.exceptions.PublicationSourceInvalidity( relative,
187 'entry absent' )
188 _check_entry_digest(
189 target, relative, get( 'sha256', '' ), semantics )
190 _check_entry_fragments(
191 target, relative, get( 'fragments', [ ] ) )
194def _check_entry_digest(
195 target: __.Path, relative: str, digest: str, semantics: str
196) -> None:
197 ''' Checks entry content digest for frozen or pinned entries. '''
198 import hashlib
199 if digest or semantics == 'frozen':
200 if not digest:
201 raise __.exceptions.PublicationSourceInvalidity(
202 relative, 'frozen entry without digest' )
203 actual = hashlib.sha256( target.read_bytes( ) ).hexdigest( )
204 if actual != digest:
205 raise __.exceptions.PublicationSourceInvalidity(
206 relative, 'content drift detected' )
209def _check_entry_fragments(
210 target: __.Path, relative: str, fragment_source: __.typx.Any
211) -> None:
212 ''' Checks entry anchor fragments as parsed element IDs. '''
213 if isinstance( fragment_source, str ):
214 raise __.exceptions.PublicationSourceInvalidity(
215 relative, 'malformed fragments' )
216 try: fragments = list( fragment_source )
217 except TypeError as exception:
218 raise __.exceptions.PublicationSourceInvalidity(
219 relative, 'malformed fragments' ) from exception
220 identifiers = extract_element_ids(
221 target.read_text( errors = 'replace' ) )
222 for fragment in fragments:
223 if not isinstance( fragment, str ) or fragment not in identifiers:
224 raise __.exceptions.PublicationSourceInvalidity( relative,
225 'fragment absent' )
228def extract_element_ids( content: str ) -> set[ str ]:
229 ''' Extracts element IDs from HTML content with a real parser. '''
230 import html.parser
231 identifiers: set[ str ] = set( )
233 class Collector( html.parser.HTMLParser ):
235 def handle_starttag(
236 self, tag: str, attrs: list[ tuple[ str, str | None ] ]
237 ) -> None:
238 for name, value in attrs:
239 if name == 'id' and value:
240 identifiers.add( value )
242 parser = Collector( convert_charrefs = True )
243 parser.feed( content )
244 return identifiers