Coverage for sources / mimeogram / fsprotect / cache.py: 95%
96 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-02-18 17:27 +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''' Cache for filesystem protection checks. '''
24from . import __
25from . import core as _core
28_scribe = __.produce_scribe( __name__ )
31class Rule( __.immut.DataclassObject ):
32 ''' Rule for path protection. '''
34 paths: frozenset[ __.Path ]
35 patterns: frozenset[ str ] = frozenset( )
38class Cache( _core.Protector ):
39 ''' Cache of protected paths and patterns for platform. '''
41 rules: dict[ _core.Reasons, Rule ]
42 defaults_disablement: frozenset[ str ]
43 rules_supercession: __.immut.Dictionary[
44 __.Path, tuple[ frozenset[ str ], frozenset[ str ] ] ]
46 @classmethod
47 def from_configuration(
48 selfclass,
49 auxdata: __.appcore.state.Globals,
50 ) -> __.typx.Self:
51 ''' Initializes protection cache for current platform. '''
52 _scribe.debug( 'Initializing protection cache.' )
53 rules: dict[ _core.Reasons, Rule ] = { }
54 discover_platform_locations( auxdata, rules )
55 provide_credentials_locations( rules )
56 provide_project_locations( rules )
57 disables, supercedes = _process_configuration( auxdata, rules )
58 return selfclass(
59 rules = rules,
60 defaults_disablement = disables,
61 rules_supercession = supercedes )
63 def verify( self, path: __.Path ) -> _core.Status:
64 ''' Verifies if a path should be protected using cached data. '''
65 path = _normalize_path( path )
66 _scribe.debug( f"Path: {path}" )
67 if any( part in self.defaults_disablement for part in path.parts ):
68 return _core.Status( path = path, active = False )
69 for dir_path, ( ignore, protect ) in self.rules_supercession.items( ):
70 dir_path_ = _normalize_path( dir_path )
71 if not path.is_relative_to( dir_path_ ): continue
72 rel_path = path.relative_to( dir_path_ )
73 if _check_path_patterns( rel_path, ignore ):
74 return _core.Status( path = path, active = False )
75 if _check_path_patterns( rel_path, protect ):
76 return _core.Status(
77 path = path,
78 reason = _core.Reasons.PlatformSensitive,
79 active = True )
80 for reason, rule in self.rules.items( ):
81 for protected_path in rule.paths:
82 protected_path_ = _normalize_path( protected_path )
83 if path.is_relative_to( protected_path_ ):
84 return _core.Status(
85 path = path,
86 reason = reason,
87 active = True )
88 if _check_path_patterns( path, rule.patterns ):
89 return _core.Status(
90 path = path,
91 reason = reason,
92 active = True )
93 return _core.Status( path = path, active = False )
96def provide_credentials_locations(
97 rules: dict[ _core.Reasons, Rule ]
98) -> None:
99 ''' Provides common locations for credentials and other secrets. '''
100 from . import home as module
101 home = __.Path.home( )
102 cred_paths = {
103 home / path for path in module.discover_sensitive_locations( ) }
104 rules[ _core.Reasons.Credentials ] = Rule(
105 paths = frozenset( cred_paths ) )
108def provide_project_locations(
109 rules: dict[ _core.Reasons, Rule ]
110) -> None:
111 ''' Provides IDE and VCS locations relative to project. '''
112 from . import project as module
113 project_sensitive = module.discover_sensitive_locations( )
114 # TODO: Consider whether these patterns are compatible with Windows.
115 rules[ _core.Reasons.Concealment ] = Rule(
116 paths = frozenset( ),
117 patterns = frozenset(
118 f"**/{path}/**" for path in project_sensitive ) )
121def _check_path_patterns( path: __.Path, patterns: frozenset[ str ] ) -> bool:
122 ''' Checks if path matches any of the glob patterns. '''
123 from wcmatch import glob
124 str_path = str( path )
125 return glob.globmatch( str_path, list( patterns ), flags = glob.GLOBSTAR )
128def discover_platform_locations(
129 auxdata: __.appcore.state.Globals,
130 rules: dict[ _core.Reasons, Rule ],
131) -> None:
132 ''' Discovers system and user locations based on platform. '''
133 match __.sys.platform:
134 case 'darwin':
135 from . import macos as module
136 sys_paths = module.discover_system_paths( )
137 user_paths = module.discover_user_paths( )
138 case 'win32':
139 from . import windows as module
140 sys_paths = module.discover_system_paths( )
141 user_paths = module.discover_user_paths( )
142 case _:
143 from . import unix as module
144 sys_paths = module.discover_system_paths( )
145 user_paths: frozenset[ __.Path ] = frozenset( )
146 rules[ _core.Reasons.OsDirectory ] = Rule( paths = frozenset( sys_paths ) )
147 config_paths = (
148 user_paths | { __.Path( auxdata.directories.user_config_path ), } )
149 rules[ _core.Reasons.UserConfiguration ] = Rule(
150 paths = frozenset( config_paths ) )
153def _expand_location( path: str ) -> __.Path:
154 ''' Expands path with home directory and environment variables. '''
155 expanded = __.os.path.expanduser( __.os.path.expandvars( path ) )
156 if ( __.sys.platform == 'win32' 156 ↛ 160line 156 didn't jump to line 160 because the condition on line 156 was never true
157 and expanded.startswith( '/' )
158 and not expanded.startswith( '//' ) # Skip UNC paths
159 ):
160 expanded = expanded.lstrip( '/' )
161 path_obj = __.Path( expanded )
162 if not path_obj.is_absolute( ):
163 path_obj = __.Path( __.Path.cwd( ).drive + '/' + expanded )
164 else: path_obj = __.Path( expanded )
165 return _normalize_path( path_obj.resolve( ) )
168def _normalize_path( path: __.Path ) -> __.Path:
169 ''' Normalizes path for consistent comparison across platforms. '''
170 resolved = path.resolve( )
171 if __.sys.platform == 'win32' and resolved.drive:
172 return __.Path(
173 resolved.drive.lower( )
174 + str( resolved )[ len( resolved.drive ): ] )
175 return resolved
178def _process_configuration(
179 auxdata: __.appcore.state.Globals,
180 rules: dict[ _core.Reasons, Rule ],
181) -> tuple[
182 frozenset[ str ],
183 __.immut.Dictionary[
184 __.Path, tuple[ frozenset[ str ], frozenset[ str ] ] ],
185]:
186 config = auxdata.configuration.get( 'protection', { } )
187 if not config: return frozenset( ), __.immut.Dictionary( )
188 # Additional locations and patterns.
189 locations_add = {
190 _expand_location( path )
191 for path in config.get( 'additional-locations', ( ) ) }
192 patterns_add = set( config.get( 'additional-patterns', ( ) ) )
193 rules[ _core.Reasons.CustomAddition ] = Rule(
194 paths = frozenset( locations_add ),
195 patterns = frozenset( patterns_add ) )
196 # Override defaults.
197 patterns_remove = frozenset( config.get( 'defaults-disablement', ( ) ) )
198 # Process directory overrides
199 supercession_rules: dict[
200 __.Path, tuple[ frozenset[ str ], frozenset[ str ] ] ] = { }
201 for dir_path, dir_rules in config.get(
202 'rules-supercession', { }
203 ).items( ):
204 full_path = _expand_location( dir_path )
205 supercession_rules[ full_path ] = (
206 frozenset( dir_rules.get( 'ignore', [ ] ) ),
207 frozenset( dir_rules.get( 'protect', [ ] ) ) )
208 return patterns_remove, __.immut.Dictionary( supercession_rules )