# Copyright 2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""Runtime access to configured logical storage volumes.
The service resolves literal steering roots in the caller's path context,
validates persistent volume identities and mediates filesystem access.
:author: Bulgheroni Antonio
.. versionadded:: 2.4
"""
from __future__ import annotations
import os
import re
import stat
import uuid
from collections.abc import Mapping, Sequence
from copy import deepcopy
from dataclasses import dataclass
from datetime import UTC, datetime
from pathlib import Path, PurePosixPath
from types import MappingProxyType
from typing import IO, TYPE_CHECKING, Any, cast
from peewee import Database, OperationalError
from mafw.storage.logical_reference import LogicalFileReference
from mafw.tools.path_resolution import (
PathResolutionContext,
PathResolutionError,
PathValidationPolicy,
resolve_path,
)
if TYPE_CHECKING:
from mafw.steering.models import (
StorageAccess,
StorageConfiguration,
StorageVolumeConfiguration,
VolumeStorageType,
)
_VOLUME_NAME_PATTERN = re.compile(r'[A-Za-z0-9_-]+\Z', re.ASCII)
_MARKER_PREFIX = '.mafw_'
_MARKER_SUFFIX = '_uuid'
_STORAGE_VOLUME_MODEL_NAME = 'StorageVolume'
[docs]
class StorageError(Exception):
"""Base exception for storage-service failures.
.. versionadded:: 2.4
"""
[docs]
class StorageConfigurationError(StorageError):
"""A configured volume cannot be used by the filesystem service.
.. versionadded:: 2.4
"""
[docs]
class StorageInitializationError(StorageError):
"""A configured volume root or persistent identity could not be validated.
.. versionadded:: 2.4
"""
[docs]
class StorageRecoveryError(StorageInitializationError):
"""Registration failed and compensating cleanup also failed.
.. versionadded:: 2.4
"""
[docs]
class StorageVolumeNotFoundError(StorageError, KeyError):
"""A reference or metadata request names an unconfigured volume.
.. versionadded:: 2.4
"""
[docs]
@dataclass(frozen=True, slots=True)
class StorageVolumeInfo:
"""Resolved runtime metadata for one validated volume.
:param name: Case-sensitive logical volume name.
:param configured_root: Literal root expression from steering configuration.
:param root: Resolved canonical filesystem root.
:param type: Configured storage backend.
:param access: MAFw access policy for service-managed operations.
:param parameters: Read-only view of backend-specific settings.
:param uuid: Validated persistent volume identity as a ``uuid.UUID``.
.. versionadded:: 2.4
"""
name: str
configured_root: str
root: Path
type: VolumeStorageType
access: StorageAccess
parameters: Mapping[str, Any]
uuid: uuid.UUID
[docs]
class StorageService:
"""Resolve logical references and access initialized storage volumes.
Construction is lightweight. Call :meth:`initialize` after the caller has
configured the MAFw Peewee database and created the ``StorageVolume`` table.
Initialization can register a new volume by creating its UUID marker and
database row. This registration write is administrative identity setup and
is allowed for any logical access policy when OS permissions permit it.
The ``read-only`` policy still prevents data-file writes through this
service.
:param configuration: Parsed storage configuration with literal root values.
:param path_context: Runtime context used to resolve configured roots.
:param environment: Resolved environment mapping for root interpolation.
.. versionadded:: 2.4
"""
def __init__(
self,
configuration: StorageConfiguration,
path_context: PathResolutionContext,
environment: Mapping[str, str],
) -> None:
self._configuration = configuration
self._path_context = path_context
self._environment = dict(environment)
self._volumes: dict[str, StorageVolumeInfo] = {}
self._initialized = False
self._volume_model: Any | None = None
@property
def initialized(self) -> bool:
"""Whether all configured volumes passed initialization.
.. versionadded:: 2.4
"""
return self._initialized
[docs]
def initialize(self) -> None:
"""Resolve and validate all volumes, registering new identities safely.
Existing markers and database identities are never modified when a
volume is unavailable or mismatched. Successfully created registrations
are durable even if a later configured volume fails; a subsequent call
validates those records as existing volumes.
:raises StorageInitializationError: If database, root, marker, or
identity validation fails.
:raises StorageConfigurationError: If configuration is unusable.
.. versionadded:: 2.4
"""
self._initialized = False
self._volumes = {}
self._volume_model = None
if self._configuration.invalid_section is not None or self._configuration.invalid_volumes:
raise StorageConfigurationError('StorageConfiguration contains malformed volume entries.')
volume_model, db = self._require_database()
self._volume_model = volume_model
pending: dict[str, StorageVolumeInfo] = {}
for name, config in self._configuration.volumes.items():
pending[name] = self._initialize_volume(name, config, db)
try:
with db.atomic():
now = datetime.now(UTC).replace(tzinfo=None)
for name in pending:
updated = volume_model.update(last_seen=now).where(volume_model.volume == name).execute()
if updated != 1:
raise StorageInitializationError(
f'Volume {name!r} disappeared before its last_seen timestamp could be updated.'
)
except StorageInitializationError:
raise
except Exception as exc:
raise StorageInitializationError('Could not update last_seen for validated storage volumes.') from exc
self._volumes = pending
self._initialized = True
[docs]
def get_volume(self, name: str) -> StorageVolumeInfo:
"""Return validated immutable runtime metadata for a configured volume.
:param name: Case-sensitive logical volume name.
:return: Resolved volume metadata.
:raises RuntimeError: If initialization has not completed.
:raises StorageVolumeNotFoundError: If the name is not configured.
.. versionadded:: 2.4
"""
self._require_initialized()
try:
return self._volumes[name]
except KeyError as exc:
raise StorageVolumeNotFoundError(name) from exc
[docs]
def resolve(self, reference: LogicalFileReference | str) -> Path:
"""Resolve a logical reference to a contained physical path.
The target need not exist, allowing callers to prepare output paths.
Existing symlinks are followed only when their targets remain inside
the effective volume root. This path-based check cannot prevent a
symlink replacement between validation and a later filesystem call.
:param reference: A validated reference or its serialized string.
:return: Canonical contained path, possibly not yet present.
:raises StorageError: If the service is not initialized or resolution
would escape the configured root.
.. versionadded:: 2.4
"""
self._require_initialized()
logical = LogicalFileReference.from_string(reference) if isinstance(reference, str) else reference
if not isinstance(logical, LogicalFileReference):
raise TypeError('reference must be a LogicalFileReference or serialized reference string')
volume = self.get_volume(logical.volume)
return self._contained_path(volume, Path(*logical.relative_path.parts))
[docs]
def open(self, reference: LogicalFileReference | str, mode: str = 'r', **kwargs: Any) -> IO[Any]:
"""Open a referenced file while enforcing its logical access policy.
``full`` and ``protected`` permit standard Python file modes, including
overwrite and append. ``read-only`` permits modes without ``+``, ``w``,
``a`` or ``x``. The service does not create parent directories. Text
access defaults to UTF-8; callers may pass other ``open`` keyword
arguments. ``protected`` has no deletion API in this public surface.
:param reference: A validated reference or serialized reference string.
:param mode: Standard Python file mode.
:param kwargs: Additional arguments forwarded to ``Path.open()``.
:return: An open file object.
:raises PermissionError: If the mode violates the volume policy.
:raises OSError: Filesystem errors, with their original cause preserved
when contextualized by path validation.
.. versionadded:: 2.4
"""
from mafw.steering.models import StorageAccess
volume = self._volume_for_reference(reference)
writes = self._mode_writes(mode)
if volume.access is StorageAccess.READ_ONLY and writes:
raise PermissionError(f'Volume {volume.name!r} is read-only; mode {mode!r} can modify data.')
path = self.resolve(reference)
if 'b' not in mode and 'encoding' not in kwargs:
kwargs['encoding'] = 'utf-8'
try:
return path.open(mode, **kwargs)
except OSError as exc:
raise OSError(
exc.errno, f'Cannot open reference {reference!s} on volume {volume.name!r}: {exc}', path
) from exc
[docs]
def delete(self, reference: LogicalFileReference | str | Sequence[LogicalFileReference | str]) -> None:
"""Delete one or more referenced files on fully accessible volumes.
Each reference is validated before unlinking. A missing target is
idempotent success; other filesystem and storage failures are raised.
For a sequence, all references are attempted and failures are reported
together so callers can retain their queue entry and retry safely.
:param reference: One logical reference or an ordered sequence.
:raises PermissionError: If any referenced volume is not ``full``.
:raises ExceptionGroup: If one or more sequence items cannot be deleted.
.. versionadded:: 2.4
Added policy-aware deletion for logical storage references.
"""
from mafw.steering.models import StorageAccess
references: Sequence[LogicalFileReference | str]
if isinstance(reference, (LogicalFileReference, str)):
references = [reference]
single = True
else:
references = reference
single = False
failures: list[Exception] = []
for item in references:
try:
volume = self._volume_for_reference(item)
if volume.access is not StorageAccess.FULL:
raise PermissionError(
f'Volume {volume.name!r} has {volume.access.value!r} access; deletion requires full access.'
)
path = self.resolve(item)
try:
path.unlink()
except FileNotFoundError:
# The volume was validated first; a genuinely absent file is already cleaned.
pass
except Exception as exc:
if single:
raise
failures.append(exc)
if failures:
raise ExceptionGroup('Could not delete all logical file references.', failures)
[docs]
def generate_reference(self, path: Path | str, *, volume: str | None = None) -> LogicalFileReference:
"""Create a logical reference for an absolute path within one volume.
:param path: Absolute path, which may identify a not-yet-created output.
:param volume: Optional explicit volume when root membership is
ambiguous.
:return: A normalized portable logical reference.
:raises StorageError: If the path is relative, outside the selected
root, or ambiguous across configured roots.
.. versionadded:: 2.4
"""
self._require_initialized()
physical = Path(path).expanduser()
if not physical.is_absolute():
raise ValueError('generate_reference() requires an absolute filesystem path')
if volume is not None:
selected = self.get_volume(volume)
return self._reference_for_path(selected, physical)
matches: list[tuple[StorageVolumeInfo, Path]] = []
for info in self._volumes.values():
try:
matches.append((info, self._contained_absolute_path(info, physical)))
except StorageConfigurationError:
continue
if not matches:
raise StorageVolumeNotFoundError(f'Path {str(physical)!r} is not within a configured volume.')
if len(matches) > 1:
names = ', '.join(info.name for info, _ in matches)
raise StorageConfigurationError(f'Path {str(physical)!r} matches multiple volumes: {names}.')
return self._reference_from_contained(matches[0][0], matches[0][1])
[docs]
def _require_database(self) -> tuple[Any, Database]:
"""Find the registered StorageVolume model and require its active DB."""
from mafw.db.db_model import get_db, mafw_model_register
try:
volume_model = cast(Any, mafw_model_register.get_model(_STORAGE_VOLUME_MODEL_NAME))
except KeyError as exc:
raise StorageInitializationError('StorageVolume is not registered with MAFw.') from exc
try:
# The registry returns Peewee's dynamic model class; get_db uses its _meta binding.
db = get_db(volume_model)
except (AttributeError, TypeError) as exc:
raise StorageInitializationError('The MAFw database is not configured for StorageVolume.') from exc
if not isinstance(db, Database):
raise StorageInitializationError('StorageVolume is not bound to an initialized Peewee database.')
if db.is_closed():
raise StorageInitializationError('The MAFw database must be connected before StorageService.initialize().')
try:
table_exists = db.table_exists(volume_model._meta.table_name)
except (OperationalError, AttributeError) as exc:
raise StorageInitializationError('Could not inspect the StorageVolume table.') from exc
if not table_exists:
raise StorageInitializationError('The StorageVolume standard table must exist before initialization.')
return volume_model, db
[docs]
def _initialize_volume(self, name: str, config: StorageVolumeConfiguration, db: Database) -> StorageVolumeInfo:
"""Resolve one root, validate/register identity, and build its view."""
from mafw.steering.models import StorageAccess, StorageVolumeConfiguration, VolumeStorageType
volume_model = self._require_volume_model()
if not isinstance(name, str) or _VOLUME_NAME_PATTERN.fullmatch(name) is None:
raise StorageConfigurationError(f'Invalid storage volume name {name!r}; expected ASCII [A-Za-z0-9_-]+.')
if not isinstance(config, StorageVolumeConfiguration):
raise StorageConfigurationError(f'Volume {name!r} configuration must be a StorageVolumeConfiguration.')
try:
storage_type = VolumeStorageType(config.type)
except (TypeError, ValueError) as exc:
raise StorageConfigurationError(f'Volume {name!r} uses unsupported backend {config.type!r}.') from exc
try:
access = StorageAccess(config.access)
except (TypeError, ValueError) as exc:
raise StorageConfigurationError(f'Volume {name!r} has invalid access policy {config.access!r}.') from exc
if storage_type is not VolumeStorageType.FILESYSTEM:
raise StorageConfigurationError(f'Volume {name!r} uses unsupported backend {storage_type.value!r}.')
if config.root is None or not isinstance(config.root, str) or not config.root:
raise StorageConfigurationError(f'Volume {name!r} must have a non-empty filesystem root.')
if config.invalid_fields:
raise StorageConfigurationError(f'Volume {name!r} contains invalid configured fields.')
if not isinstance(config.parameters, Mapping):
raise StorageConfigurationError(f'Volume {name!r} parameters must be a mapping.')
parameters = config.parameters.get(storage_type.value, {})
if not isinstance(parameters, Mapping):
raise StorageConfigurationError(f'Volume {name!r} parameters must be a mapping.')
try:
root = resolve_path(
config.root,
self._path_context,
validation=PathValidationPolicy.EXISTING_DIRECTORY,
env=self._environment,
)
except (PathResolutionError, OSError) as exc:
raise StorageInitializationError(f'Cannot resolve root for volume {name!r}: {exc}') from exc
marker = root / f'{_MARKER_PREFIX}{name}{_MARKER_SUFFIX}'
try:
record = volume_model.get_or_none(volume_model.volume == name)
except Exception as exc:
raise StorageInitializationError(f'Cannot query identity for volume {name!r}.') from exc
if record is None:
identity = self._register_new_volume(name, marker, db)
else:
identity = self._read_marker(name, marker)
if identity != record.uuid:
raise StorageInitializationError(
f'Volume {name!r} marker UUID {identity!r} does not match database UUID {record.uuid!r}.'
)
self._validate_unique_identity(name, identity)
return StorageVolumeInfo(
name=name,
configured_root=config.root,
root=root,
type=storage_type,
access=access,
parameters=MappingProxyType(deepcopy(dict(parameters))),
uuid=uuid.UUID(identity),
)
[docs]
def _register_new_volume(self, name: str, marker: Path, db: Database) -> str:
"""Create a new marker exclusively, then insert its DB identity."""
volume_model = self._require_volume_model()
if marker.is_symlink() or marker.exists():
raise StorageInitializationError(
f'Volume {name!r} has a marker but no database registration; explicit recovery is required.'
)
identity = str(uuid.uuid4())
marker_created = False
try:
self._create_marker_exclusively(marker, identity)
marker_created = True
with db.atomic():
volume_model.create(volume=name, uuid=identity)
except FileExistsError as exc:
record = volume_model.get_or_none(volume_model.volume == name)
if record is not None:
existing = self._read_marker(name, marker)
if existing == record.uuid:
self._validate_unique_identity(name, existing)
return existing
raise StorageInitializationError(
f'Another initializer created marker for volume {name!r}; identity pair is incomplete or mismatched.'
) from exc
except Exception as exc:
if marker_created:
try:
self._remove_created_marker(marker, identity)
except Exception as cleanup_exc:
raise StorageRecoveryError(
f'Registration of volume {name!r} failed and its newly created marker could not be removed; '
f'verify marker {str(marker)!r} and database registration manually.'
) from cleanup_exc
if isinstance(exc, StorageError):
raise
raise StorageInitializationError(f'Could not register new volume {name!r}.') from exc
return identity
[docs]
def _create_marker_exclusively(self, marker: Path, identity: str) -> None:
"""Write a UUID marker without following or replacing an existing file."""
flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
if hasattr(os, 'O_NOFOLLOW'):
flags |= os.O_NOFOLLOW
descriptor = os.open(marker, flags, 0o600)
created_stat: os.stat_result | None = None
try:
created_stat = os.fstat(descriptor)
with os.fdopen(descriptor, 'w', encoding='utf-8') as handle:
handle.write(f'{identity}\n')
handle.flush()
os.fsync(handle.fileno())
except Exception as exc:
try:
os.close(descriptor)
except OSError:
pass
try:
current_stat = marker.lstat()
if (
created_stat is not None
and stat.S_ISREG(current_stat.st_mode)
and (
current_stat.st_dev,
current_stat.st_ino,
)
== (created_stat.st_dev, created_stat.st_ino)
):
marker.unlink()
else:
raise OSError(f'Marker {str(marker)!r} changed during creation.')
except Exception as cleanup_exc:
raise StorageRecoveryError(
f'Writing marker {str(marker)!r} failed and its partial file could not be removed.'
) from cleanup_exc
raise StorageInitializationError(f'Could not write marker {str(marker)!r}.') from exc
[docs]
def _remove_created_marker(self, marker: Path, identity: str) -> None:
"""Compensate only if this attempt's regular marker is still unchanged."""
if marker.is_symlink() or not marker.is_file():
raise OSError(f'Marker {str(marker)!r} changed before cleanup.')
if marker.read_text(encoding='utf-8') != f'{identity}\n':
raise OSError(f'Marker {str(marker)!r} no longer contains the created UUID.')
marker.unlink()
[docs]
def _read_marker(self, name: str, marker: Path) -> str:
"""Read a canonical UUID4 marker without following a symlink."""
descriptor: int | None = None
try:
flags = os.O_RDONLY
if hasattr(os, 'O_NOFOLLOW'):
flags |= os.O_NOFOLLOW
descriptor = os.open(marker, flags)
info = os.fstat(descriptor)
if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode):
raise StorageInitializationError(f'Marker for volume {name!r} must be a regular, non-symlink file.')
with os.fdopen(descriptor, 'rb') as handle:
descriptor = None
content = handle.read().decode('utf-8')
except FileNotFoundError as exc:
raise StorageInitializationError(f'Marker for registered volume {name!r} is missing.') from exc
except UnicodeError as exc:
raise StorageInitializationError(f'Marker for volume {name!r} is not valid UTF-8.') from exc
except OSError as exc:
raise StorageInitializationError(f'Cannot read marker for volume {name!r}.') from exc
finally:
if descriptor is not None:
os.close(descriptor)
value = content[:-1] if content.endswith('\n') else content
try:
parsed = uuid.UUID(value)
except (ValueError, AttributeError) as exc:
raise StorageInitializationError(f'Marker for volume {name!r} does not contain a valid UUID4.') from exc
if parsed.version != 4 or str(parsed) != value or content not in {value, f'{value}\n'}:
raise StorageInitializationError(
f'Marker for volume {name!r} must contain canonical lowercase hyphenated UUID4 text.'
)
return value
[docs]
def _validate_unique_identity(self, name: str, identity: str) -> None:
"""Ensure a database UUID does not identify another logical volume."""
volume_model = self._require_volume_model()
duplicate = volume_model.get_or_none((volume_model.uuid == identity) & (volume_model.volume != name))
if duplicate is not None:
raise StorageInitializationError(
f'UUID for volume {name!r} is also registered to volume {duplicate.volume!r}.'
)
[docs]
def _require_volume_model(self) -> Any:
"""Return the model resolved from MAFw's registry for this initialization."""
if self._volume_model is None:
raise StorageInitializationError('StorageVolume is not available before database initialization.')
return self._volume_model
[docs]
def _volume_for_reference(self, reference: LogicalFileReference | str) -> StorageVolumeInfo:
"""Return initialized volume metadata for a reference."""
self._require_initialized()
logical = LogicalFileReference.from_string(reference) if isinstance(reference, str) else reference
if not isinstance(logical, LogicalFileReference):
raise TypeError('reference must be a LogicalFileReference or serialized reference string')
return self.get_volume(logical.volume)
[docs]
def _contained_path(self, volume: StorageVolumeInfo, relative: Path) -> Path:
"""Resolve a relative target and reject symlink escape or dangling links."""
candidate = volume.root / relative
return self._check_containment(volume, candidate)
[docs]
def _contained_absolute_path(self, volume: StorageVolumeInfo, candidate: Path) -> Path:
"""Resolve an absolute candidate and require it to remain in one root."""
return self._check_containment(volume, candidate)
[docs]
def _check_containment(self, volume: StorageVolumeInfo, candidate: Path) -> Path:
"""Canonicalize an output-capable path and verify root containment."""
try:
resolved = candidate.resolve(strict=False)
resolved.relative_to(volume.root)
self._reject_dangling_symlinks(volume.root, candidate)
except ValueError as exc:
raise StorageConfigurationError(
f'Path {str(candidate)!r} escapes volume {volume.name!r} root {str(volume.root)!r}.'
) from exc
except (OSError, RuntimeError) as exc:
raise StorageInitializationError(
f'Cannot validate path {str(candidate)!r} on volume {volume.name!r}.'
) from exc
return resolved
[docs]
@staticmethod
def _reject_dangling_symlinks(root: Path, candidate: Path) -> None:
"""Reject any existing symlink component that cannot be resolved strictly."""
try:
relative = candidate.absolute().relative_to(root)
except ValueError:
return
current = root
for part in relative.parts:
current = current / part
if current.is_symlink():
current.resolve(strict=True)
[docs]
def _reference_for_path(self, volume: StorageVolumeInfo, path: Path) -> LogicalFileReference:
"""Apply the same canonical containment rules for explicit-volume paths."""
return self._reference_from_contained(volume, self._contained_absolute_path(volume, path))
[docs]
@staticmethod
def _reference_from_contained(volume: StorageVolumeInfo, path: Path) -> LogicalFileReference:
"""Convert a checked physical path to its portable relative reference."""
relative = path.relative_to(volume.root)
if not relative.parts:
raise ValueError('A storage root cannot be represented as a file reference.')
return LogicalFileReference(volume.name, PurePosixPath(*relative.parts))
[docs]
@staticmethod
def _mode_writes(mode: str) -> bool:
"""Validate Python mode syntax and report whether it can mutate data."""
if not isinstance(mode, str) or not mode:
raise ValueError('file mode must be a non-empty string')
if any(character not in 'rwaxbt+' for character in mode):
raise ValueError(f'invalid file mode: {mode!r}')
bases = [character for character in mode if character in 'rwax']
if len(bases) != 1 or mode.count('+') > 1 or mode.count('b') > 1 or mode.count('t') > 1:
raise ValueError(f'invalid file mode: {mode!r}')
if 'b' in mode and 't' in mode:
raise ValueError(f'invalid file mode: {mode!r}')
return bases[0] in 'wax' or '+' in mode
[docs]
def _require_initialized(self) -> None:
"""Reject runtime operations until the complete volume set is valid."""
if not self._initialized:
raise RuntimeError('StorageService.initialize() must complete before runtime access.')