Index: CENSE =================================================================== --- LICENSE (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,7 +1,0 @@ -* Lizenzhinweis - -Teile der Software Flex-O-Grader wurden während der Dienstausübung mit Genehmigung entwickelt. -Daher können der Bundeswehr interne Nutzungsrechte an diesen Bestandteilen zustehen. - -Bis zur abschließenden Klärung der Nutzungs- und Veröffentlichungsrechte -unterliegt eine Weitergabe oder Veröffentlichung der Zustimmung der Dienststelle. Index: README.md =================================================================== --- README.md (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ README.md (revision d6e75be17bbd83e4a95dcabd999d3cefa40bc8db) @@ -1,587 +1,32 @@ +# flexoentity -# Table of Contents +**flexoentity** is the foundational Python library of the *Flex-O* ecosystem. -- [Overview](#org77ab3e0) -- [Flex-O ID Structure](#orgf136db1) -- [Lifecycle States](#orgea1c2ca) -- [Core Classes](#org2b0320a) - - [FlexOID](#org39ec589) - - [`FlexoEntity`](#orgd84c86d) -- [Integrity Verification](#org4c3e14b) -- [Real World Example](#orgfb82c02) -- [Usage](#orge03e624) -- [Serialization Example](#org118d77d) -- [Entity Type and State Codes](#org83e21be) -- [Design Notes](#orga31954b) -- [Dependencies](#org5589bef) -- [Integration](#org7b599dd) -- [License](#org56e2d0f) +It provides: +- `FlexOID`: a deterministic, state-aware, and versioned identifier system +- `FlexoEntity`: a base class for reproducible, lifecycle-tracked domain entities +Together, these components ensure traceability, immutability, and auditability +across the Flex-O-Grader, Flex-O-Vault, and Flex-O-Drill projects. +### Example - +```python +from flexoentity import FlexOID, FlexoEntity -# Overview +question = FlexoEntity( + domain="AF", + etype="Q", + text="What is Ohm’s law?", + state="A" +) +print(question.flexo_id) -\`flexoentity\` provides the **identity and lifecycle backbone** for all Flex-O components -(Flex-O-Grader, Flex-O-Vault, Flex-O-Drill, …). +Features + • BLAKE2s hashing (fast, stdlib, deterministic) + • Canonical seed generation + • Lifecycle state encoding (@001A) + • UTC-based timestamps + • Collision disambiguation -It defines how entities such as questions, media, catalogs, and exams are **identified, versioned, signed, and verified** — all without any external database dependencies. - -At its heart lie two modules: - -- `id_factory.py` – robust, cryptographically-verifiable **Flex-O ID generator** -- `flexo_entity.py` – abstract **base class for all versioned entities** - -Together, they form a compact yet powerful foundation for audit-ready, reproducible data structures across offline and air-gapped deployments. - -- Design Goals - - - - --- -- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
GoalDescription
DeterminismIDs are derived from canonicalized entity content — identical input always yields identical ID prefix.
IntegrityBLAKE2s hashing and digital signatures protect against manual tampering.
TraceabilityVersion numbers (@001A, @002S, …) track entity lifecycle transitions.
StabilityHash prefixes remain constant across state changes; only version and state suffixes evolve.
AuditabilityEvery entity can be serialized, verified, and reconstructed without hidden dependencies.
SimplicityPure-Python, zero external libraries, self-contained and easy to embed.
- - - - -# Flex-O ID Structure - -Each entity carries a unique **Flex-O ID**, generated by `FlexOID.generate()`. - - AF-I250101-9A4C2D@003S - - - - --- -- -- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
SegmentExampleMeaning
DomainAF or PY_LANGUppercase - Logical scope (e.g. “Air Force”)
TypeIEntity type (e.g. ITEM)
Date250101UTC creation date (YYMMDD)
Hash9A4C2D4F6E5312-digit BLAKE2s digest of canonical content
Version003Sequential version counter
StateSLifecycle state (D, A, S, P, O)
- - - - -# Lifecycle States - - - - --- -- -- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
StateAbbrev.Description
DRAFTDEditable, not yet validated
APPROVEDAReviewed and accepted
APPROVEDANDSIGNEDSCryptographically signed
PUBLISHEDPReleased to consumers
OBSOLETEOArchived or replaced
- -Transitions follow a strict progression: - - DRAFT -> APPROVED -> APPROVED_AND_SIGNED -> PUBLISHED -> OBSOLETE - -Only DRAFT entities can be deleted - all others got OBSOLETE mark instead - - - - -# Core Classes - - - - -## FlexOID - -A lightweight immutable class representing the full identity of an entity. - -**Highlights** - -- safegenerate(domain, entitytype, estate, text, version=1, repo) -> create a new ID -- nextversion(oid) -> increment version safely -- clonenewbase(domain, entitytype, estate, text) -> start a new lineage -- Deterministic prefix, state-dependent signature - - - - -## `FlexoEntity` - -Abstract base class for all versioned entities (e.g., Question, Exam, Catalog). - -Implements: - -- ID lifecycle management (approve(), sign(), publish(), obsolete()) -- Serialization (tojson(), fromjson(), todict(), fromdict()) -- Integrity verification (verifyintegrity(entity)) -- Controlled state transitions with automatic timestamps - -Subclasses define a single property: - - @property - def text_seed(self) -> str: - """Canonical text or core content for hashing.""" - - - - -# Integrity Verification - -Each entity can self-verify its integrity: - - entity = Question.with_domain_id(domain_id="AF", text="What is Ohm’s law?", topic="Electronics") - json_str = entity.to_json() - reloaded = Question.from_json(json_str) - - assert FlexoEntity.verify_integrity(reloaded) - -If the file is tampered with (e.g. “Ohm’s” → “Omm’s”), verification fails: - - - - -# Real World Example - -Below you can see the implementation of a dedicated FlexoEntity class, used for Domains. -We set an ENTITYTYPE and define the needed fields in the data class. We define how to create -a default object, the textseed (it is easy because the domain id is unique and therefore sufficient) -and the methods for serialization. - - from uuid import UUID - from dataclasses import dataclass - from flexoentity import FlexOID, FlexoEntity, EntityType - - @dataclass - class Domain(FlexoEntity): - """ - I am a helper class to provide more information than just a - domain abbreviation in FlexOID, doing mapping and management - """ - - ENTITY_TYPE = EntityType.DOMAIN - - fullname: str = "" - description: str = "" - classification: str = "UNCLASSIFIED" - - @classmethod - def default(cls): - """Return the default domain object.""" - return cls.with_domain_id(domain_id="GEN_GENERIC", - fullname="Generic Domain", classification="UNCLASSIFIED") - - @property - def text_seed(self) -> str: - return self.domain_id - - def to_dict(self): - base = super().to_dict() - base.update({ - "flexo_id": self.flexo_id, - "domain_id": self.domain_id, - "fullname": self.fullname, - "description": self.description, - "classification": self.classification, - }) - return base - - @classmethod - def from_dict(cls, data): - # Must have flexo_id - if "flexo_id" not in data: - raise ValueError("Domain serialization missing 'flexo_id'.") - - flexo_id = FlexOID(data["flexo_id"]) - - obj = cls( - fullname=data.get("fullname", ""), - description=data.get("description", ""), - classification=data.get("classification", "UNCLASSIFIED"), - flexo_id=flexo_id, - _in_factory=True - ) - - # Restore metadata - obj.origin = data.get("origin") - obj.fingerprint = data.get("fingerprint", "") - obj.originator_id = ( - UUID(data["originator_id"]) if data.get("originator_id") else None - ) - obj.owner_id = ( - UUID(data["owner_id"]) if data.get("owner_id") else None - ) - - return obj - - - - -# Usage - - d = Domain.default() - print(d.flexo_id) # GEN_GENERIC-D251124-67C2CAE292CE@001D - d.approve() - print(d.flexo_id) # GEN_GENERIC-D251124-67C2CAE292CE@001A - d.sign() - print(d.flexo_id) # GEN_GENERIC-D251124-67C2CAE292CE@001S - - - - -# Serialization Example - - { - 'flexo_id': FlexOID(GEN_GENERIC-D251124-29CE0F4BE59D@001S), - 'fingerprint': '534BD2EC5C5511F1', - 'origin': FlexOID(GEN_GENERIC-D251124-67C2CAE292CE@001D), - 'originator_id': '00000000-0000-0000-0000-000000000000', - 'owner_id': '00000000-0000-0000-0000-000000000000', - 'domain_id': 'GEN_GENERIC', - 'fullname': 'Generic Domain', - 'description': '', - 'classification': 'UNCLASSIFIED'} - - { - "flexo_id": "GEN_GENERIC-D251124-29CE0F4BE59D@001S", - "fingerprint": "534BD2EC5C5511F1", - "origin": "GEN_GENERIC-D251124-67C2CAE292CE@001D", - "originator_id": "00000000-0000-0000-0000-000000000000", - "owner_id": "00000000-0000-0000-0000-000000000000", - "domain_id": "GEN_GENERIC", - "fullname": "Generic Domain", - "description": "", - "classification": "UNCLASSIFIED" - } - - - - -# Entity Type and State Codes - - - - --- -- -- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
EntityTypeCodeTypical Use
GENERICGGeneric entities that does not fit other types yet or are temporarily only
DOMAINDEvery Domain is of this type
MEDIAMEvery media item belongs to this type, e.g. Pictures, Audio, Video
ITEMIAn Entity what is usually used in a collection, e.g. Questions in a test
COLLECTIONCA collection of items, as an Exam or a catalog
TEXTTA text document
HANDOUTHA published document
OUTPUTOThe output of a computation
RECORDRRecord type data, as bibliography entries
SESSIONSA unique session, e.g. managed by a session manager
USERUUser objects
CONFIGFCONFIG files that need to be tracked over time and state
EVENTEEvents that have to be tracked over time, as status messages or orders
ATTESTATIONXEntities that attest a formal technical (not human) check e.g. Signatures
- - - - --- -- -- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
EntityStateCodeMeaning
DRAFTDWork in progress
APPROVEDAReviewed
APPROVEDANDSIGNEDSSigned version
PUBLISHEDPPublicly released
OBSOLETEODeprecated
- - - - -# Design Notes - -- **Hash Stability:** Only domain, entity type, and content text influence the hash. - This ensures consistent prefixes across state changes. -- **State-Dependent Signatures:** Each lifecycle stage has its own signature seed. - Modifying a file without re-signing invalidates integrity. -- **Obsolescence Threshold:** Version numbers above 900 trigger warnings; - beyond 999 are considered obsolete. -- **Clone Lineages:** Cloning an entity resets versioning but preserves metadata lineage. - - - - -# Dependencies - -- Python 3.11+ -- Standard library only (`hashlib`, `json`, `datetime`, `enum`, `dataclasses`) - -No external packages. Fully compatible with **Guix**, **air-gapped** deployments, and **reproducible builds**. - - - - -# Integration - -\`flexoentity\` is imported by higher-level modules such as: - -- **Flex-O-Grader** → manages question catalogs and exam bundles -- **Flex-O-Vault** → provides persistent media storage with metadata integrity -- **Flex-O-Drill** → uses versioned entities for training simulations - -All share the same identity and versioning logic — ensuring that -**what was approved, signed, and published remains provably authentic.** - - - - -# License - -MIT License 2025 -Part of the **Flex-O family** by Flex-O-Dyne GmbH -Designed for reproducible, audit-ready, human-centered software. - +Licensed under the MIT License. Index: flexoentity/__init__.py =================================================================== --- flexoentity/__init__.py (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ flexoentity/__init__.py (revision d6e75be17bbd83e4a95dcabd999d3cefa40bc8db) @@ -6,50 +6,8 @@ - FlexoEntity: lifecycle-tracked base class for all Flex-O domain objects """ -# FIXME: change the default log path to /var/log/ammos/ammos.log when system -# configuration created this dir with the appropriate rights -from importlib.metadata import version, PackageNotFoundError -from .logger import logger from .id_factory import FlexOID, canonical_seed -from .flexo_entity import FlexoEntity, EntityType, EntityState -from .flexo_collection import FlexoCollection -from .typed_collection import TypedCollection -from .in_memory_backend import InMemoryBackend -from .runtime_backend import RuntimeBackend -from .composite_backend import CompositeBackend -from .entity_manager import EntityManager -from .sqlite_entity_backend import SQLiteEntityBackend -from .domain import Domain -from .domain_manager import DomainManager, DuplicateDomainError -from .flexo_signature import FlexoSignature, CertificateReference -from .entity_registry import EntityRegistry -from .signing_backends import get_signing_backend +from .flexo_entity import FlexoEntity -__all__ = [ - "FlexOID", - "canonical_seed", - "FlexoEntity", - "EntityType", - "Domain", - "EntityManager", - "SQLiteEntityBackend", - "RuntimeBackend", - "InMemoryBackend", - "CompositeBackend", - "DomainManager", - "DuplicateDomainError", - "EntityState", - "FlexoCollection", - "TypedCollection", - "FlexoSignature", - "EntityRegistry", - "CertificateReference", - "get_signing_backend", - "logger" -] - -# Optional: keep dynamic version retrieval synced with pyproject.toml -try: - __version__ = version("flexoentity") -except PackageNotFoundError: - __version__ = "0.0.0-dev" +__all__ = ["FlexOID", "canonical_seed", "FlexoEntity"] +__version__ = "1.0.0" Index: exoentity/composite_backend.py =================================================================== --- flexoentity/composite_backend.py (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,110 +1,0 @@ -from .persistance_backend import PersistanceBackend -from .flexo_entity import FlexoEntity - - -class CompositeBackend(PersistanceBackend): - """ - Backend wrapper. - - Option A semantics: - - Reads come from the primary backend only. - - Writes propagate to primary and all sync backends. - - All backends store/return dicts. - """ - - def __init__(self, authoritative_backend, sync_backends=None): - # Validate entity_class existence and compatibility - entity_class = getattr(authoritative_backend, "entity_class", None) - if entity_class is None: - raise TypeError("primary_backend must expose .entity_class") - - if not issubclass(entity_class, FlexoEntity): - raise TypeError("entity_class must be a subclass of FlexoEntity") - - super().__init__(entity_class) - - self._primary = authoritative_backend - self.sync_backends = list(sync_backends or []) - - for b in self.sync_backends: - if b.entity_class != self._primary.entity_class: - raise TypeError( - f"Backend {b} does not match entity_class={self.entity_class.__name__}" - ) - - @property - def primary(self): - return self._primary - - def add_sync_backend(self, backend, clear=False): - if backend.entity_class != self.primary.entity_class: - raise TypeError("Backend entity_class mismatch") - - if clear: - backend.clear() - - # Sync current data into backend - for d in self.primary.load_all(): - backend.save(d) - - self.sync_backends.append(backend) - - def remove_backend(self, backend): - if backend is self.primary: - raise ValueError("Cannot remove the primary backend") - self.sync_backends.remove(backend) - - # --------------------------------------------------------- - # Write operations propagate to *all* backends (dicts) - # --------------------------------------------------------- - - def save(self, entity_dict: dict): - self.primary.save(entity_dict) - for b in self.sync_backends: - b.save(entity_dict) - - def update(self, entity_dict: dict): - self.primary.update(entity_dict) - for b in self.sync_backends: - b.update(entity_dict) - - def delete(self, flexo_id: str): - self.primary.delete(flexo_id) - for b in self.sync_backends: - b.delete(flexo_id) - - # --------------------------------------------------------- - # Read operations from primary only - # --------------------------------------------------------- - - def load(self, flexo_id: str): - return self.primary.load(flexo_id) - - def load_all(self): - return self.primary.load_all() - - # --------------------------------------------------------- - # Sync helpers - # --------------------------------------------------------- - - def sync_all(self, clear_targets=False): - """ - Push all data from primary backend to the other backends. - If clear_targets=True, wipe sync backends first. - """ - if clear_targets: - for b in self.sync_backends: - b.clear() - - for d in self.primary.load_all(): - for b in self.sync_backends: - b.save(d) - - def clear(self): - self.primary.clear() - for b in self.sync_backends: - b.clear() - - def __repr__(self): - names = ", ".join(b.__class__.__name__ for b in self.sync_backends) - return f"" Index: flexoentity/domain.py =================================================================== --- flexoentity/domain.py (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ flexoentity/domain.py (revision d6e75be17bbd83e4a95dcabd999d3cefa40bc8db) @@ -1,48 +1,22 @@ -from flexoentity import FlexoEntity, EntityType - +@dataclass class Domain(FlexoEntity): - - ENTITY_TYPE = EntityType.DOMAIN - - def __init__(self, *, flexo_id, fullname="", description="", subtype=None, fingerprint=None, - classification="UNCLASSIFIED", origin=None, originator_id=None, owner_id=None,): - - self.fullname = fullname - self.description = description - self.classification = classification - - super().__init__( - flexo_id=flexo_id, - subtype=subtype, - fingerprint=fingerprint, - origin=origin, - originator_id=originator_id, - owner_id=owner_id, - ) - - @classmethod - def default(cls): - """Return the default domain object.""" - return cls.with_domain_id(domain_id="GEN_GENERIC", - fullname="Generic Domain", classification="UNCLASSIFIED") + abbrev: str + fullname: str + description: str = "" + classification: str = "UNCLASSIFIED" + owner: str = "unknown" @property - def text_seed(self) -> str: - return self.domain_id + def qtype(self): # or entity_type + return EntityType.DOMAIN - def _deserialize_content(self, content: dict): - super()._deserialize_content(content) - self.fullname = content.get("fullname", "") - self.description = content.get("description", "") - self.classification = content.get("classification", "UNCLASSIFIED") - - def _serialize_content(self): - base = super()._serialize_content() - base.update( - { - "fullname": self.fullname, - "description": self.description, - "classification": self.classification, - } - ) + def to_dict(self): + base = super().to_dict() + base.update({ + "abbrev": self.abbrev, + "fullname": self.fullname, + "description": self.description, + "classification": self.classification, + "owner": self.owner + }) return base Index: exoentity/domain_manager.py =================================================================== --- flexoentity/domain_manager.py (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,136 +1,0 @@ -from __future__ import annotations -import json - -from .flexo_collection import FlexoCollection -from .domain import Domain -from .flexo_entity import EntityState -from .entity_manager import EntityManager - - -class DomainNotFoundError(KeyError): - pass - - -class InvalidDomainError(ValueError): - pass - - -class DuplicateDomainError(ValueError): - pass - - -class DomainManager(EntityManager): - """ - Manager for Domain entities using the new backend architecture. - - - Primary storage: backend.load_all() (InMemory, SQLite, JSON, ...) - - Secondary index: domain_id → Domain - - Domain-specific business rules included. - """ - - ENTITY_CLASS = Domain - - def __init__(self, local_backend, staging_backend, permanent_backend, registry=None): - assert local_backend is not None - assert staging_backend is not None - assert permanent_backend is not None - super().__init__(local_backend=local_backend, - staging_backend=staging_backend, - permanent_backend=permanent_backend) - self._registry = registry - self._index_by_domain_id = {} - - self._rebuild_index() - - def refresh(self): - self._rebuild_index - - # ------------------------------------------------------------ - # Index rebuild - # ------------------------------------------------------------ - - def _rebuild_index(self): - """Rebuild domain_id lookup table from backend contents.""" - self._index_by_domain_id.clear() - for backend in [self.local_backend, self.staging_backend, self.permanent_backend]: - for dom in backend.load_all(): - self._index_by_domain_id[dom.domain_id] = dom - - # ------------------------------------------------------------ - # Overrides for CRUD - # ------------------------------------------------------------ - - def add(self, domain: Domain): - # Domain-specific uniqueness rule - if domain.domain_id in self._index_by_domain_id: - raise DuplicateDomainError(f"Domain '{domain.domain_id}' already exists.") - - super().add(domain) - self._index_by_domain_id[domain.domain_id] = domain - - def update(self, domain: Domain): - super().update(domain) - self._index_by_domain_id[domain.domain_id] = domain - - def delete(self, flexo_id: str): - dom = self.local_backend.load(flexo_id) - if dom is None: - return - - domain_id = dom.domain_id - - # Check domain deletion rules - if not self.can_delete(domain_id): - raise ValueError(f"Domain '{domain_id}' is still referenced.") - - super().delete(flexo_id) - - # Remove from index - self._index_by_domain_id.pop(domain_id, None) - - # ------------------------------------------------------------ - # Domain-specific lookup - # ------------------------------------------------------------ - - def get_by_domain_id(self, domain_id: str) -> Domain: - dom = self._index_by_domain_id.get(domain_id) - if not dom: - raise KeyError(f"Domain '{domain_id}' not found.") - return dom - - def find_by_domain_id(self, domain_id: str): - return self._index_by_domain_id.get(domain_id) - - def all_domain_ids(self): - return list(self._index_by_domain_id.keys()) - - def all(self): - return list(self._index_by_domain_id.values()) - # ------------------------------------------------------------ - # Domain deletion rules - # ------------------------------------------------------------ - - def can_delete(self, domain_id: str) -> bool: - if not self._registry: - return True - return not any(self._registry.entities_by_domain(domain_id)) - - # ------------------------------------------------------------ - # Domain lifecycle helpers - # ------------------------------------------------------------ - - def ensure_approved(self, domain_id: str) -> None: - dom = self.get_by_domain_id(domain_id) - - if dom.state != EntityState.APPROVED: - raise ValueError( - f"Domain '{domain_id}' must be APPROVED before " - f"entities in this domain can be promoted. " - f"(Current state: {dom.state})" - ) - - def clear(self): - self.local_backend.clear() - - def __repr__(self): - return f"" Index: exoentity/entity_manager.py =================================================================== --- flexoentity/entity_manager.py (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,172 +1,0 @@ -from .flexo_entity import FlexoEntity -from .typed_collection import TypedCollection - - -class EntityNotFoundError(KeyError): - pass - - -class InvalidEntityError(ValueError): - pass - - -class DuplicateEntityError(ValueError): - pass - - -class EntityManager: - """ - Backend-agnostic manager for any FlexoEntity subclass. - - Option A contract: - - backends store/return dicts only - - manager converts dict <-> entity - """ - - ENTITY_CLASS = None # must be overridden by subclasses - - def __init__(self, local_backend, staging_backend, permanent_backend): - if self.ENTITY_CLASS is None: - raise ValueError("Subclasses must define ENTITY_CLASS") - - if not issubclass(self.ENTITY_CLASS, FlexoEntity): - raise TypeError("ENTITY_CLASS must be a subclass of FlexoEntity") - - self.local_backend = local_backend - self.staging_backend = staging_backend - self.permanent_backend = permanent_backend - - # ------------------------------------------------------------------ - # Conversion helpers - # ------------------------------------------------------------------ - - def _to_dict(self, entity) -> dict: - self._ensure_type(entity) - return entity.to_dict() - - def _to_entity(self, entity_dict: dict): - if entity_dict is None: - return None - return self.ENTITY_CLASS.from_dict(entity_dict) - - def _to_entities(self, dicts: list[dict]): - return [self._to_entity(d) for d in dicts] - - # ------------------------------------------------------------------ - # Locators - # ------------------------------------------------------------------ - - def backend_of_domain_id(self, domain_id: str): - # Note: this converts dict->entity because domain_id is a FlexoEntity property - for backend_name, backend in [ - ("local", self.local_backend), - ("staging", self.staging_backend), - ("permanent", self.permanent_backend), - ]: - for d in backend.load_all(): - e = self._to_entity(d) - if e.domain_id == domain_id: - return backend_name - return None - - def backend_of_flexo_id(self, flexo_id: str): - for backend_name, backend in [ - ("local", self.local_backend), - ("staging", self.staging_backend), - ("permanent", self.permanent_backend), - ]: - if backend.load(flexo_id) is not None: - return backend_name - return None - - # ------------------------------------------------------------------ - # CRUD operations (entity API) - # ------------------------------------------------------------------ - - def add(self, entity): - self.local_backend.save(self._to_dict(entity)) - - def update(self, entity): - self.local_backend.update(self._to_dict(entity)) - - def delete(self, flexo_id: str): - self.local_backend.delete(flexo_id) - - # ------------------------------------------------------------------ - # Retrieval (entity API) - # ------------------------------------------------------------------ - - def get(self, flexo_id: str): - d = self.local_backend.load(flexo_id) - return self._to_entity(d) - - # FIXME: Readd staging backend later - def all(self): - dicts = ( - self.local_backend.load_all() - # + self.staging_backend.load_all() - + self.permanent_backend.load_all() - ) - return self._to_entities(dicts) - - def promote_to_staging(self, flexo_id): - d = self.local_backend.load(flexo_id) - if d is None: - raise EntityNotFoundError(flexo_id) - self.staging_backend.save(d) - self.local_backend.delete(flexo_id) - - def promote_to_permanent(self, flexo_id): - d = self.staging_backend.load(flexo_id) - if d is None: - raise EntityNotFoundError(flexo_id) - self.permanent_backend.save(d) - self.staging_backend.delete(flexo_id) - - def sync_all(self): - # sync staging → local - for d in self.staging_backend.load_all(): - self.local_backend.save(d) - - # sync permanent → local - for d in self.permanent_backend.load_all(): - self.local_backend.save(d) - - # NOTE: refresh() is not defined here. - # If you want a hook, define it explicitly, otherwise remove this call. - if hasattr(self, "refresh"): - self.refresh() - - # ------------------------------------------------------------------ - # Helpers - # ------------------------------------------------------------------ - - def exists(self, flexo_id: str) -> bool: - return self.local_backend.load(flexo_id) is not None - - def count(self) -> int: - return len(self.local_backend.load_all()) - - def add_or_update(self, entity): - fid = str(entity.flexo_id) - if self.exists(fid): - self.update(entity) - else: - self.add(entity) - - def as_collection(self): - # Collection expects entities (your current TypedCollection usage suggests this) - return TypedCollection(self.ENTITY_CLASS, items=self.all()) - - # ------------------------------------------------------------------ - # Internal - # ------------------------------------------------------------------ - - def _ensure_type(self, entity): - if not isinstance(entity, self.ENTITY_CLASS): - raise TypeError( - f"Expected {self.ENTITY_CLASS.__name__}, got {type(entity).__name__}" - ) - - def __repr__(self): - return f"" Index: exoentity/entity_registry.py =================================================================== --- flexoentity/entity_registry.py (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,26 +1,0 @@ -class EntityRegistry: - """ - Central authority for accessing all FlexoEntities. - Works with in-memory managers today, - and can be extended to SQLite later. - """ - - def __init__(self): - self._sources = [] # list of callables - - def register_source(self, func): - """Register a callable that returns a list or iterable of entities.""" - if not callable(func): - raise TypeError("Source must be callable") - self._sources.append(func) - - def all_entities(self): - """Iterate over all known entities.""" - for src in self._sources: - for e in src(): - yield e - - def entities_by_domain(self, domain_id): - for e in self.all_entities(): - if e.flexo_id.domain_id == domain_id: - yield e Index: exoentity/flexo_collection.py =================================================================== --- flexoentity/flexo_collection.py (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,134 +1,0 @@ -""" -flexoentity.collection ----------------------- - -A minimal collection for FlexOEntities. - -- Default key = entity.flexo_id -- Optional override key=... for special cases (e.g., domains) -- Smalltalk-like protocol preserved -- Dict-based: fast lookup -- Backwards compatible API -""" -from abc import abstractmethod -from .flexo_entity import FlexoEntity -from .id_factory import FlexOID - - -class FlexoCollection: - """ - A minimal collection of FlexOEntities, keyed by FlexOID by default. - - Examples: - coll = FlexoCollection() - coll.add(entity) # uses entity.flexo_id as key - coll.add(domain, key=domain.domain_id) # override key if needed - ent = coll.get(some_id) - for e in coll: - ... - """ - - def __init__(self, items=None): - # internal dict-based storage - self._items = {} - - if items: - for it in items: - self.add(it) - - # ------------------------------------------------------------------ - # Core API (unchanged externally, enhanced internally) - # ------------------------------------------------------------------ - - def add(self, entity: FlexoEntity, key=None): - """ - Add or replace an entity. - - - If key is None → use entity.flexo_id - - If key is given → use caller-provided key - """ - if key is None: - try: - key = entity.flexo_id - except AttributeError: - raise TypeError( - "Items must have .flexo_id or explicitly use add(entity, key=...)." - ) - - self._items[key] = entity # overwrite is intentional & preserved - - def remove(self, oid: FlexOID): - """Remove an entity by ID, ignoring if missing (API preserved).""" - self._items.pop(oid, None) - - def clear(self): - self._items = {} - - def get(self, oid: FlexOID): - """Return the entity or None.""" - return self._items.get(oid) - - def __contains__(self, oid: FlexOID): - """Check if an ID exists.""" - return oid in self._items - - def __len__(self) -> int: - return len(self._items) - - def __iter__(self): - return iter(self._items.values()) - - # ------------------------------------------------------------------ - # Additional public helpers (unchanged) - # ------------------------------------------------------------------ - def entities(self): - """Return all entities as a list.""" - return list(self._items.values()) - - def ids(self): - """Return all stored keys.""" - return list(self._items.keys()) - - # ------------------------------------------------------------------ - # Smalltalk-inspired interface (unchanged) - # ------------------------------------------------------------------ - - def at(self, oid): - """Smalltalk-style accessor (raises if missing).""" - return self._items[oid] - - def at_put(self, oid, entity): - """Smalltalk-style setter (kept).""" - self._items[oid] = entity - - def includes_id(self, oid: FlexOID): - """Smalltalk-style membership test.""" - return oid in self._items - - def do(self, block): - """Smalltalk-style #do: iteration.""" - for value in self._items.values(): - block(value) - - # ------------------------------------------------------------------ - # Serialization helpers (unchanged) - # ------------------------------------------------------------------ - - def serialize(self): - """Serialize all entities to list of dicts.""" - return [e.to_dict() for e in self.entities()] - - @classmethod - @abstractmethod - def deserialize(cls, data): - """ - Deserialize a list of dicts into a FlexoCollection. - - Uses from_dict() to dispatch to the correct subclass. - """ - pass - - # ------------------------------------------------------------------ - - def __repr__(self) -> str: - return f"" Index: flexoentity/flexo_entity.py =================================================================== --- flexoentity/flexo_entity.py (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ flexoentity/flexo_entity.py (revision d6e75be17bbd83e4a95dcabd999d3cefa40bc8db) @@ -1,70 +1,15 @@ """ -flexo_entity.py - I represent the mutable counterpart of the immutable FlexOID. - -I exist to connect an entity’s immutable identity with its mutable provenance -and content. My FlexOID encodes what an entity *is* - its domain, type, -version, and lifecycle state - while I record *who* created it, *who* owns it -now, and *where* it came from. - -Structure -────────── -- My `flexo_id` is the canonical source of truth. - It deterministically encodes: - - entity type (single letter, e.g. I, C, M) - - date + hash (prefix stability) - - version (numeric counter) - - state (single capital letter) - -- My derived properties (state, entity_type, version) read from that ID. - They are never stored separately, so I cannot become inconsistent. - -- My attributes origin, originator_id, and owner_id express authorship - and workflow context: - - origin links to the FlexOID of the entity I was derived from - - originator_id identifies the original creator (UUID) - - owner_id identifies the current responsible user or process (UUID) - -Design principles -────────────────── -I separate immutable identity from mutable context: - - ┌────────────────────────────┐ - │ FlexOID │ - │ – who I am │ - │ – lifecycle state │ - └────────────▲───────────────┘ - │ - ┌────────────┴───────────────┐ - │ FlexoEntity (this module) │ - │ – who created or owns me │ - │ – where I came from │ - │ – what content I hold │ - └────────────────────────────┘ - -This means: - - My identity never changes; only my content or ownership does. - - My lifecycle transitions (approve, sign, publish) are represented by new - FlexOIDs, not by mutating my own attributes. - - My audit trail and access control live in the provenance layer, not the ID. - -I am the living object behind a FlexOID — mutable, traceable, and accountable. +versioned_entity.py — Shared base class for all Flex-O entities. +Integrates with hardened id_factory.py and content fingerprint tracking. """ - import json -from uuid import UUID -from enum import Enum -from dataclasses import dataclass, field, replace +from enum import Enum, auto +from dataclasses import dataclass, field +from datetime import datetime from typing import Optional -from pathlib import Path -from abc import ABC, abstractmethod -import hashlib - - -from flexoentity.id_factory import FlexOID -from flexoentity.logger import logger - - -SCHEMA_VERSION = "1.0" -SCHEMA_NAME = "flexograder-entity" +from abc import ABC + +from id_factory import FlexOID + # ────────────────────────────────────────────────────────────────────────────── @@ -74,307 +19,148 @@ class EntityType(Enum): - """ - I map a fixed number of entity types to a single unique letter, - that structures different types - """ - GENERIC = "G" - DOMAIN = "D" - MEDIA = "M" - ITEM = "I" - CATALOG = "C" - TEXT = "T" - HANDOUT = "H" - OUTPUT = "O" - RECORD = "R" - SESSION = "S" - USER = "U" - CONFIG = "F" - EVENT = "E" - ATTESTATION = "X" + QUESTION = auto() + MEDIA = auto() + CATALOG = auto() + EXAM = auto() + DATABASE = auto() + CERTIFICATE = auto() + + def short(self) -> str: + mapping = { + EntityType.QUESTION: "Q", + EntityType.CATALOG: "CAT", + EntityType.EXAM: "EX", + EntityType.DATABASE: "DB", + EntityType.CERTIFICATE: "CERT" + } + return mapping[self] + +# ────────────────────────────────────────────────────────────────────────────── +# States +# ────────────────────────────────────────────────────────────────────────────── + + +class EntityState(Enum): + DRAFT = auto() + APPROVED = auto() + APPROVED_AND_SIGNED = auto() + PUBLISHED = auto() + OBSOLETE = auto() + + def short(self) -> str: + """ + Return a one-letter abbreviation for the state, used in Flex-O IDs. + """ + mapping = { + EntityState.DRAFT: "D", + EntityState.APPROVED: "A", + EntityState.APPROVED_AND_SIGNED: "S", + EntityState.OBSOLETE: "O", + } + return mapping[self] @classmethod - def from_letter(cls, a_letter): - """I am a convenience constructor method for EntityType(a_letter)""" - return cls(a_letter) - -# ────────────────────────────────────────────────────────────────────────────── -# States -# ────────────────────────────────────────────────────────────────────────────── - - -class EntityState(Enum): - """I describe the life cycle states for all entity types""" - DRAFT = "D" - APPROVED = "A" - APPROVED_AND_SIGNED = "S" - PUBLISHED = "P" - OBSOLETE = "O" - - def allowed_transitions(self) -> list[str]: - """Return allowed state names for this state.""" - return TRANSITIONS[self] + def from_short(cls, char: str): + """ + Inverse of .short(): restore the EntityState from its one-letter code. + """ + reverse = { + "D": cls.DRAFT, + "A": cls.APPROVED, + "S": cls.APPROVED_AND_SIGNED, + "O": cls.OBSOLETE, + } + try: + return reverse[char.upper()] + except KeyError: + raise ValueError(f"Unknown state abbreviation: {char}") def __str__(self): - return self.name - - -TRANSITIONS = { - EntityState.DRAFT: [EntityState.APPROVED], - EntityState.APPROVED: [EntityState.APPROVED_AND_SIGNED, EntityState.PUBLISHED, - EntityState.OBSOLETE], - EntityState.APPROVED_AND_SIGNED: [EntityState.PUBLISHED, EntityState.OBSOLETE], - EntityState.PUBLISHED: [EntityState.OBSOLETE], - EntityState.OBSOLETE: [], -} - - + return self.value + + +@dataclass class FlexoEntity(ABC): - """ - I represent a mutable entity identified by an immutable FlexOID. - - My flexo_id is the single source of truth for what I am: - - it encodes my domain, entity type, version, and lifecycle state. - I never store those values separately, but derive them on demand. - - I extend the raw identity with provenance and authorship data: - - origin -> the FlexOID of the entity I was derived from - - originator_id -> the UUID of the first creator - - owner_id -> the UUID of the current responsible person or system - - I am designed to be traceable and reproducible: - - My identity (flexo_id) never mutates. - - My lifecycle transitions are represented by new FlexOIDs. - - My provenance (origin, originator_id, owner_id) can change over time - as the entity moves through workflows or ownership transfers. - - Behavior - ───────── - - state and entity_type are read-only views derived from my FlexOID. - - to_dict() and from_dict() serialize me with human-readable fields - while preserving the canonical FlexOID. - - Subclasses (such as questions, media items, or catalogs) extend me with - domain-specific content and validation, but not with new identity rules. - - Subclasses have to define a class attribute ENTITY_TYPE to allow specific instance creation - - I am the living, editable layer that connects identity with accountability. - """ - - def with_new_owner(self, new_owner: UUID): - """I return a clone of this entity with a different owner.""" - copy = replace(self) - copy.owner_id = new_owner - return copy - - def __init__(self, *, flexo_id, subtype=None, fingerprint=None, - origin=None, originator_id=None, owner_id=None): - - if flexo_id is None: - raise ValueError("flexo_id must be provided") - - self.flexo_id = flexo_id - self.subtype = subtype or self.__class__.__name__ - self.origin = origin - self.originator_id = originator_id - self.owner_id = owner_id - self.fingerprint = fingerprint - - @staticmethod - def canonicalize_content_dict(data) -> str: - """ - Canonicalize structured content for fingerprinting. - - Stable JSON representation - - Sorted keys - - No whitespace noise - """ - return json.dumps(data, sort_keys=True, separators=(",", ":")) - - def __eq__(self, other): - if not isinstance(other, FlexoEntity): - return NotImplemented - return self.flexo_id == other.flexo_id - - def __hash__(self): - return hash(self.flexo_id) - - @property - @abstractmethod - def text_seed(self) -> str: - """I provide a canonicalized text used for ID generation.""" - raise NotImplementedError("Subclasses must define text_seed property") - - @property - def domain_id(self): - """I return my domain_id derived from FlexOID""" - return self.flexo_id.domain_id - - @property - def state(self) -> EntityState: - """I return the state derived from my FlexOID""" - return EntityState(self.flexo_id.state_code) - - @property - def entity_type(self) -> EntityType: - """I return the entity type derived from my FlexOID""" - return EntityType(self.flexo_id.entity_type) - - @classmethod - @abstractmethod - def default(cls): - """I return a minimal valid instance of this entity (in DRAFT state).""" - raise NotImplementedError("Subclasses must implement default()") - - - @classmethod - def with_domain_id(cls, domain_id: str, **kwargs): - entity_type = getattr(cls, "ENTITY_TYPE", None) - if not entity_type: - raise ValueError(f"{cls.__name__} must define ENTITY_TYPE") - - oid = FlexOID.safe_generate( - domain_id=domain_id, - entity_type=entity_type.value, - state=EntityState.DRAFT.value, - text="", - version=1, - ) - - obj = cls(flexo_id=oid, **kwargs) - obj.fingerprint = obj._compute_fingerprint() - return obj + domain: str + etype: EntityType + text_seed: str + state: EntityState = EntityState.DRAFT + flexo_id: FlexOID = field(init=False) + created_at: datetime = field(default_factory=datetime.utcnow) + updated_at: Optional[datetime] = None + + # ─────────────────────────────────────────────────────────────── + + def __post_init__(self): + """Generate ID and content fingerprint.""" + self.flexo_id = FlexOID.generate(self.domain, self.etype.short(), self.state.short(), self.text_seed, 1) def __str__(self): return ( - f"{self.subtype}-{self.entity_type.name}({self.flexo_id}, {self.state.name}, " - f"fingerprint={self.fingerprint}..., v{self.version})" + f"{self.etype.name}({self.flexo_id}, {self.state.name}, " + f"sig={self.flexo_id.signature}..., v{self.version})" ) - def meta_dict(self): + def to_dict(self): return { + "domain": self.domain, + "etype": self.etype.name, + "text_seed": self.text_seed, + "state": self.state.name, + "version": self.version, "flexo_id": str(self.flexo_id), - "subtype": self.subtype, - "fingerprint": self.fingerprint, - "origin": self.origin, - "originator_id": str(self.originator_id), - "owner_id": str(self.owner_id), - } - - def to_dict(self): - - return { - "meta": { - "schema": { - "name": SCHEMA_NAME, - "version": SCHEMA_VERSION, - }, - **self.meta_dict(), - }, - "content": self._serialize_content(), + "signature": self.flexo_id.signature, + "created_at": self.created_at.isoformat(), + "updated_at": self.updated_at.isoformat() if self.updated_at else None, } @classmethod - def from_dict(cls, data: dict): - meta = data["meta"] - content = data["content"] - - flexo_id = FlexOID(meta["flexo_id"]) - + def from_dict(cls, data): obj = cls( - flexo_id=flexo_id, - subtype=meta.get("subtype"), - fingerprint=meta.get("fingerprint"), - origin=meta.get("origin"), - originator_id=meta.get("originator_id"), - owner_id=meta.get("owner_id"), + data["domain"], + EntityType[data["etype"]], + data["text_seed"], + EntityState[data["state"]], + data["version"], ) - - obj._deserialize_content(content) - + obj.flexo_id = FlexOID(data["flexo_id"], data.get("signature", "")) + if data.get("updated_at"): + obj.updated_at = datetime.fromisoformat(data["updated_at"]) return obj - def fork( - self, - *, - domain_id: str | None = None, - reset_state: bool = True, - keep_origin: bool = True, - ) -> "FlexoEntity": - """ - Create a new independent entity derived from this one. - """ - data = self.to_dict() - meta = data["meta"] - - # Provenance - if keep_origin: - meta["origin"] = str(self.flexo_id) - - # Reset lifecycle - if reset_state: - meta["version"] = 1 - meta["state"] = EntityState.DRAFT.value - - # Generate new identity - new_flexo_id = FlexOID.safe_generate( - domain_id=domain_id or self.domain_id, - entity_type=self.entity_type.value, - state=meta["state"], - text=self.text_seed, - version=meta["version"], - ) - - meta["flexo_id"] = str(new_flexo_id) - - return self.__class__.from_dict(data) - - def to_json(self, indent: int = 2) -> str: - """Return a JSON representation including all FlexO metadata.""" - return json.dumps(self.to_dict(), ensure_ascii=False, indent=indent) - - def to_json_file(self, path): - Path(path).write_text(self.to_json(), encoding="utf-8") - return path + def to_json(self, *, indent: int | None = None) -> str: + """Serialize entity (and its FlexOID) into JSON.""" + return json.dumps(self.to_dict(), indent=indent, ensure_ascii=False) @classmethod - def from_json(cls, data_str: str): - """I create a new instance from a JSON string.""" + def from_json(cls, data_str: str) -> "FlexoEntity": + """Deserialize from a JSON string.""" data = json.loads(data_str) return cls.from_dict(data) - @classmethod - def from_json_file(cls, filename: str): - with open(filename, "r", encoding="utf-8") as f: - data = json.load(f) - return cls.from_dict(data) - - - def _deserialize_content(self, content_dict): - """Subclasses override this to restore real fields.""" - pass # default: do nothing - - - def _serialize_content(self): - """ - Subclasses override to return all integrity-relevant fields. - Should include: - - all actual structured data - - comment (if meaningful) - - signature_data / certificate info (if present) - """ - return {} - - def canonical_content(self) -> str: - """Get canonical JSON string of content for fingerprinting.""" - return self.canonicalize_content_dict(self._serialize_content()) - - def _compute_fingerprint(self) -> str: - """I recompute the entity's content fingerprint.""" - return hashlib.blake2s(self.canonical_content().encode("utf-8"), - digest_size=8).hexdigest().upper() - + # ─────────────────────────────────────────────────────────────── def _update_fingerprint(self) -> bool: - new_fp = self._compute_fingerprint() - changed = new_fp != self.fingerprint - self.fingerprint = new_fp - return changed + """Recalculate fingerprint and return True if content changed.""" + # extract version from current flexo_id + new_oid = FlexOID.generate(self.domain, self.etype, self.text_seed, self.flexo_id.version) + if new_oid.signature != self.flexo_id.signature: + self.flexo_id = new_oid + return True + return False + + # ─────────────────────────────────────────────────────────────── + def _transition(self, target_state: EntityState): + """Internal helper for state transitions with version and fingerprint checks.""" + if target_state == EntityState.OBSOLETE: + self.state = target_state + return + + # Check if version should bump + if FlexOID.should_version(self.etype, target_state): + if self._update_fingerprint(): + self.flexo_id = FlexOID.next_version(self.flexo_id) + + self.state = target_state + self.updated_at = datetime.utcnow() # ─────────────────────────────────────────────────────────────── @@ -384,5 +170,5 @@ @property def version(self) -> int: - """I extract numeric version from the FlexO-ID string.""" + """Extract numeric version from the FlexO-ID string.""" try: return int(str(self.flexo_id).rsplit("@", 1)[-1]) @@ -391,150 +177,84 @@ def bump_version(self): - """I increment the version number on the ID.""" + """Increment version number on the ID.""" self.flexo_id = FlexOID.next_version(self.flexo_id) - - def lineage(self, repo): - """I return full ancestry chain [origin → ... → self].""" - chain = [self] - current = self - while current.origin and current.origin in repo: - current = repo[current.origin] - chain.insert(0, current) - return chain + self.updated_at = datetime.utcnow() def approve(self): """ - I move from DRAFT to APPROVED state. - Draft entities receive a new permanent FlexOID with incremented version. + Move from DRAFT to APPROVED state. + Draft entities receive a new permanent FlexOID. """ if self.state == EntityState.DRAFT: - new_fid = FlexOID.safe_generate(domain_id=self.domain_id, - entity_type=self.entity_type.value, - state=EntityState.APPROVED.value, - text=self.text_seed, - version=self.version - ) - self.origin = self.flexo_id # optional: keep audit trail + # Generate a brand new permanent ID + new_fid = FlexOID.generate( + self.domain, + self.etype, + self.text_seed, + draft=False + ) + self.previous_id = self.flexo_id # optional: keep audit trail self.flexo_id = new_fid + + self.state = EntityState.APPROVED + self.updated_at = datetime.utcnow() return self raise ValueError("Only drafts can be approved") def sign(self): - """ - I mark entity as approved and signed without bumping version. - Only changes the state letter in the FlexOID. - FIXME: We need a signature here - """ - if self.state != EntityState.APPROVED: - raise ValueError("Only approved entities can be signed.") - - # Change only the trailing state letter (A → S) - old_id = self.flexo_id - self.flexo_id = self.flexo_id.with_state(EntityState.APPROVED_AND_SIGNED.value) - self.origin = old_id - return self + # FIXME: We need to define clear conditions, when resigning is neccessary and allowed + # if self.state == EntityState.APPROVED: + # self._transition(EntityState.APPROVED_AND_SIGNED) + # self.bump_version() + self._transition(EntityState.APPROVED_AND_SIGNED) + self.bump_version() def publish(self): - """ - I move from APPROVED or APPROVED_AND_SIGNED to PUBLISHED. - - Uses allowed_transitions() to verify legality, - then performs a version bump and lineage update. - """ - if EntityState.PUBLISHED not in self.allowed_transitions(): - raise ValueError( - f"Illegal state transition: {self.state.name} → PUBLISHED. " - ) - - new_fid = FlexOID.safe_generate( - domain_id=self.domain_id, - entity_type=self.entity_type.value, - state=EntityState.PUBLISHED.value, - text=self.text_seed, - version=self.version - ) - - self.origin = self.flexo_id - self.flexo_id = new_fid - return self + if self.state == EntityState.APPROVED_AND_SIGNED: + self._transition(EntityState.PUBLISHED) def obsolete(self): - """I mark myself as obsolete""" - if EntityState.OBSOLETE not in self.allowed_transitions(): - raise ValueError( - f"Illegal state transition: {self.state.name} -> OBSOLETE. " - ) if self.state != EntityState.OBSOLETE: - new_fid = FlexOID.safe_generate( - domain_id=self.domain_id, - entity_type=self.entity_type.value, - state=EntityState.OBSOLETE.value, - text=self.text_seed, - version=self.version - ) - - self.origin = self.flexo_id - self.flexo_id = new_fid - - return self + self._transition(EntityState.OBSOLETE) def clone_new_base(self): """Start new lineage when obsolete.""" - self.origin = str(self.flexo_id) - self.flexo_id = FlexOID.clone_new_base( - domain_id=self.domain_id, - entity_type=self.entity_type.value, - state=EntityState.DRAFT.value, - text=self.text_seed, - ) - return self - - # ─────────────────────────────────────────────────────────────── - # Integrity verification - # ─────────────────────────────────────────────────────────────── + self.flexo_id = FlexOID.clone_new_base(self.domain, self.etype, self.text_seed) + self.state = EntityState.DRAFT + self.updated_at = datetime.utcnow() + + # ─────────────────────────────────────────────────────────────── + def modify_content(self, new_text_seed: str): + """ + Update content and detect whether fingerprint changes. + Does not increment version automatically; transitions handle that. + """ + self.text_seed = new_text_seed.strip() + if self._update_fingerprint(): + # mark content changed, but version will only bump on next approval/sign/publish + self.updated_at = datetime.utcnow() + @staticmethod def verify_integrity(entity) -> bool: """ - Verify that an entity's FlexOID, state, and fingerprint are consistent. - - Returns False if: - - flexo_id format is invalid - """ - try: - if not isinstance(entity.flexo_id, FlexOID): - return False - - # --- Fingerprint validation --- - if hasattr(entity, "fingerprint") and entity.fingerprint: - expected_fp = entity._compute_fingerprint() - if expected_fp != entity.fingerprint: - return False - - return True - - except Exception as e: - print(e) - return False + Verify if the stored fingerprint matches recalculated fingerprint. + Returns True if intact, False if tampered or corrupted. + """ + recalculated = FlexOID.generate(entity.domain, entity.etype, entity.text_seed, entity.version) + return recalculated.signature == entity.flexo_id.signature def allowed_transitions(self) -> list[str]: """Return a list of possible next state names (for GUI use).""" - return self.state.allowed_transitions() - - def apply_state_change(self, new_state): - """ - I safely apply the requested state transition by invoking - the corresponding FlexoEntity method if available. - """ - - if self.state == new_state: - return - if new_state == EntityState.APPROVED: - self.approve() - elif new_state == EntityState.APPROVED_AND_SIGNED: - self.sign() - elif new_state == EntityState.PUBLISHED: - self.publish() - elif new_state == EntityState.OBSOLETE: - self.obsolete() - else: - raise RuntimeError(f"No handler for state transition to {new_state}") + match self.state: + case EntityState.DRAFT: + return ["APPROVED"] + case EntityState.APPROVED: + return ["APPROVED_AND_SIGNED", "OBSOLETE"] + case EntityState.APPROVED_AND_SIGNED: + return ["PUBLISHED", "OBSOLETE"] + case EntityState.PUBLISHED: + return ["OBSOLETE"] + case EntityState.OBSOLETE: + return [] + case _: + return [] Index: exoentity/flexo_logging.conf =================================================================== --- flexoentity/flexo_logging.conf (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,21 +1,0 @@ -[loggers] -keys=root - -[handlers] -keys=stream_handler - -[formatters] -keys=formatter - -[logger_root] -level=DEBUG -handlers=stream_handler - -[handler_stream_handler] -class=StreamHandler -level=DEBUG -formatter=formatter -args=(sys.stderr,) - -[formatter_formatter] -format=%(asctime)s %(name)-12s %(levelname)-8s %(message)s Index: exoentity/flexo_signature.py =================================================================== --- flexoentity/flexo_signature.py (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,134 +1,0 @@ -import base64 -from dataclasses import dataclass -from typing import Optional -from uuid import UUID -from flexoentity import FlexoEntity, FlexOID, EntityType - - -@dataclass -class CertificateReference: - """ - A cross-platform reference to a signing certificate. - - This does NOT contain private keys or secret material. - It only contains the information required for OS-provided APIs - (OpenSSL, certutil, security) to *locate* or *verify* the certificate. - - Fields: - platform: - Optional override for backend creation. - Values: "LINUX", "WINDOWS", "MACOS" (case-insensitive). - If None, platform.system() determines the backend. - - identifier: - Linux: path to certificate PEM file - Windows: certificate thumbprint (SHA1, with or without colons) - macOS: certificate Common Name (CN) in Keychain - - private_key_path: - Linux only: - path to private key PEM file. - Windows/macOS: - MUST be None. - - public_cert_path: - The path to the public certificate for verification. - Linux: - optional (defaults to `identifier`) - Windows: - optional; used if you want to validate against a specific cert - macOS: - REQUIRED (used for OpenSSL verification) - """ - - platform: Optional[str] = None - identifier: str = "" - private_key_path: Optional[str] = None - public_cert_path: Optional[str] = None - - def to_dict(self): - return { - "platform": self.platform, - "identifier": self.identifier, - "private_key_path": self.private_key_path, - "public_cert_path": self.public_cert_path - } - - -class FlexoSignature(FlexoEntity): - ENTITY_TYPE = EntityType.ATTESTATION - - def __init__( - self, - *, - flexo_id, - signed_entity: Optional[FlexOID] = None, - signer_id: Optional[UUID] = None, - signature_data: str = "", - signature_type: str = "PKCS7-DER", - certificate_reference: CertificateReference | None = None, - certificate_thumbprint: str = "", - comment: Optional[str] = None, - subtype=None, - fingerprint=None, - origin=None, - originator_id=None, - owner_id=None, - ): - # content fields - self.signed_entity = signed_entity - self.signer_id = signer_id - self.signature_data = signature_data - self.signature_type = signature_type - self.certificate_reference = certificate_reference - self.certificate_thumbprint = certificate_thumbprint - self.comment = comment - - # meta fields - super().__init__( - flexo_id=flexo_id, - subtype=subtype, - fingerprint=fingerprint, - origin=origin, - originator_id=originator_id, - owner_id=owner_id, - ) - - def _serialize_content(self): - return { - "signed_entity": str(self.signed_entity) if self.signed_entity else None, - "signer_id": str(self.signer_id) if self.signer_id else None, - "signature_data": self.signature_data, - "signature_type": self.signature_type, - "certificate_thumbprint": self.certificate_thumbprint, - "comment": self.comment, - } - - @property - def text_seed(self): - return f"{self.signed_entity}:{self.signer_id}:{self.certificate_thumbprint}" - - @classmethod - def default(cls): - """ - Required by FlexoEntity. - Returns an empty draft signature. - """ - return cls.with_domain_id(domain_id="GEN") - - @classmethod - def create_signed(cls, data: bytes, entity: FlexOID, signer_id: UUID, backend): - sig = backend.sign(data) - - return cls.with_domain_id( - domain_id=entity.domain_id, - signed_entity=entity, - signer_id=signer_id, - signature_data=base64.b64encode(sig).decode(), - certificate_reference=backend.cert_ref, - certificate_thumbprint=backend.certificate_thumbprint, - ) - - def verify(self, data: bytes, backend) -> bool: - raw = base64.b64decode(self.signature_data) - return backend.verify(data, raw) Index: flexoentity/id_factory.py =================================================================== --- flexoentity/id_factory.py (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ flexoentity/id_factory.py (revision d6e75be17bbd83e4a95dcabd999d3cefa40bc8db) @@ -1,337 +1,241 @@ """ -id_factory.py — Central Flex-O ID generation and lifecycle logic. - -I represent the identity backbone of the Flex-O ecosystem. -I define how entities receive, preserve, and evolve their unique identifiers -throughout their lifecycle. - -Each Flex-O ID (FlexOID) is a self-contained string that encodes: - - --@ - -Example: - GEN-ITEM251101-3139ACFAE38B@002A - -where - DOMAIN_ID — a prefix identifying the origin domain (e.g., PY_ARITHM) - ENTITY_TYPE — a compact entity type single letter code (e.g., "I" for ITEM) - YYMMDD — the UTC creation date - HASH — a random cryptographic nonce ensuring global uniqueness - VERSION — a three-digit lineage counter (001-999) - STATE — a single capital letter indicating lifecycle state - -─────────────────────────────────────────────────────────────────────────────── -Lifecycle semantics -─────────────────────────────────────────────────────────────────────────────── - -I distinguish clearly between *content evolution* and *workflow state*: - -• Draft → Approved: - I generate a fully new FlexOID - This step marks the transition from provisional to permanent identity. - No extra version bump is required. - -• Approved → Signed → Published → Obsolete: - These transitions do not change the content hash. - They represent confidence or visibility, not content. - The same FlexOID (same version) remains valid across them. - -In short: - Version changes record evolution of content. - State changes record evolution of trust. - -─────────────────────────────────────────────────────────────────────────────── -Implementation notes -─────────────────────────────────────────────────────────────────────────────── - -• Hashing: I use modern BLAKE2s (6-byte digest) for deterministic, - collision-resistant identifiers. - -• Canonicalization: I flatten entity data into a reproducible text form - so equivalent content always yields the same hash. - -• Generation: I provide safe and deterministic factories - (FlexOID.generate, FlexOID.safe_generate) - that handle rare hash collisions by minimal salting. - -• Validation: Every FlexOID validates itself through a strict regex pattern - and exposes lightweight accessors (domain_id, type, date, hash, - version, state). - -─────────────────────────────────────────────────────────────────────────────── -Philosophy -─────────────────────────────────────────────────────────────────────────────── - -I strive to be deterministic, minimal, and auditable. -My design favors immutability and clarity over flexibility. -A FlexOID never lies about its lineage: once created, it is final. - -Version marks evolution of content. -State marks evolution of confidence. +id_factory.py — Central Flex-O ID generator and versioning control (hardened). + +Improvements: +- BLAKE2s hashing (modern, fast, stdlib) +- 6 hex-digit hash (≈16.7M combinations) +- UTC-based dates for consistency +- Collision disambiguator (-A, -B, ...) +- Canonical seed and content fingerprint helpers """ -import re +import logging +from datetime import datetime, timezone import hashlib -import secrets +import itertools import json -from datetime import datetime, timezone -from flexoentity import logger - -def canonical_seed(text: str) -> str: - """Canonicalize identity-only text_seed.""" - if not text: - return "" - # remove control characters - s = re.sub(r"[\t\r\n]+", " ", text) - # collapse multiple spaces - s = re.sub(r"\s+", " ", s) - return s.strip() - -class FlexOID(str): + +logger = logging.getLogger(__name__) +# ────────────────────────────────────────────────────────────────────────────── +# Canonicalization helpers +# ────────────────────────────────────────────────────────────────────────────── + + +def canonical_seed(obj) -> str: """ - I represent a canonical textual identifier within the Flex-O ecosystem. -i I am immutable and behave like a normal string, but I understand - my own internal structure. I can reveal my parts, create new - versions, and switch between lifecycle states. + Deterministically flatten an entity's core data into a string + for hashing and deduplication. """ - + if isinstance(obj, str): + text = " ".join(obj.lower().split()) + return text + if isinstance(obj, dict): + return json.dumps(obj, sort_keys=True, separators=(",", ":")) + if hasattr(obj, "__dict__"): + return canonical_seed(obj.__dict__) + return str(obj) + +# ────────────────────────────────────────────────────────────────────────────── +# ID Factory +# ────────────────────────────────────────────────────────────────────────────── + + +class FlexOID: MAX_VERSION = 999 WARN_THRESHOLD = 900 - OID_PATTERN = re.compile( - r"^(?P[A-Z0-9_]+)-" - r"(?P[A-Z0-9]+)" - r"(?P\d{6})-" - r"(?P[A-F0-9]+)@" - r"(?P\d{3})" - r"(?P[A-Z])$" - ) + # keep in-memory registry for same-session collisions (optional) + _seen_hashes = set() + @classmethod - def from_strings(cls, domain_id, entity_type, date, hash_part, version, state): - """ - Construct a FlexOID from raw string components only. - All parameters must already be strings in the exact expected format. - No coercions, no guesses, no defaults. - - Required formats: - domain_id: [A-Z0-9_]+ - entity_type: [A-Z] - date: YYMMDD (6 digits) - hash_part: uppercase hex, length >= 1 - version: 001..999 as 3-digit string - state: single capital letter - """ - - # Domain format - if not isinstance(domain_id, str) or not re.fullmatch(r"[A-Z0-9_]+", domain_id): - raise ValueError(f"Invalid domain_id string: {domain_id}") - - # Entity type format - if not isinstance(entity_type, str) or not re.fullmatch(r"[A-Z]", entity_type): - raise ValueError(f"Invalid entity_type string: {entity_type}") - - # Date format - if not isinstance(date, str) or not re.fullmatch(r"\d{6}", date): - raise ValueError(f"Invalid date string: {date}") - - # Hash format - if not isinstance(hash_part, str) or not re.fullmatch(r"[A-F0-9]+", hash_part): - raise ValueError(f"Invalid hash string: {hash_part}") - - # Version - if not isinstance(version, str) or not re.fullmatch(r"\d{3}", version): - raise ValueError(f"Invalid version string: {version}") - - version_int = int(version) - if not 1 <= version_int <= cls.MAX_VERSION: - raise ValueError(f"Version {version} out of range.") - - # State - if not isinstance(state, str) or not re.fullmatch(r"[A-Z]", state): - raise ValueError(f"Invalid state: {state}") - - oid_str = f"{domain_id}-{entity_type}{date}-{hash_part}@{version}{state}" - return cls(oid_str) - - @classmethod - def from_dict(cls, d): - try: - return cls.from_strings( - domain_id=d["domain_id"], - entity_type=d["entity_type"], - date=d["date"], - hash_part=d["hash"], - version=d["version"], - state=d["state"], - ) - except KeyError as e: - raise ValueError(f"Missing required FlexOID field: {e}") - - def __new__(cls, value: str): - """ - I create a new validated Flex-O ID from *value*. - I verify that the given string matches the required pattern - and remember the parsed match for later access. - """ - m = cls.OID_PATTERN.match(value) - if not m: - raise ValueError(f"Invalid FlexOID format: {value}") - obj = super().__new__(cls, value) - obj._m = m - return obj - - # ─────────────────────────────────────────── - # Parsed accessors - # ─────────────────────────────────────────── + def from_oid_and_version(cls, oid, version: int): + if not (1 <= version <= cls.MAX_VERSION): + raise ValueError(f"Version {version} out of bounds (1..{cls.MAX_VERSION}).") + return FlexOID(f"{oid.prefix}@{version:03d}{oid.state_code}", oid.signature) + + def __init__(self, flexo_id: str, signature: str): + self.flexo_id = flexo_id + self.signature = signature + + def __eq__(self, other): + if not isinstance(other, FlexOID): + return NotImplemented + return self.flexo_id == other.flexo_id and self.signature == other.signature + + def __lt__(self, other): + return self.version < other.version + + def __hash__(self): + return hash((self.flexo_id, self.signature)) + + @staticmethod + def _blake_hash(text: str) -> str: + """Return a 6-hex BLAKE2s digest.""" + return hashlib.blake2s(text.encode("utf-8"), digest_size=3).hexdigest().upper() # 3 bytes → 6 hex + + @staticmethod + def _ensure_unique(hash_part: str, version: int | None = None) -> str: + """Append disambiguator if hash was already seen this session.""" + candidate = f"{hash_part}-{version}" if version is not None else hash_part + if candidate not in FlexOID._seen_hashes: + FlexOID._seen_hashes.add(candidate) + return candidate + # fallback only if truly same hash+version (should never happen) + for suffix in itertools.chain(map(chr, range(65, 91)), range(1, 100)): + alt = f"{candidate}-{suffix}" + if alt not in FlexOID._seen_hashes: + FlexOID._seen_hashes.add(alt) + return alt + raise RuntimeError("Too many collisions; adjust hash length or logic.") + + # ────────────────────────────────────────────────────────────────────────── + @staticmethod + def generate(domain: str, etype: str, estate: str, text: str, version: int = 1): + """ + Generate a new, versioned, and state-aware Flex-O ID. + + This is the primary constructor for new entities. It combines the domain, + entity type, creation date, unique hash (derived from the canonicalized + text and lifecycle state), version number, and state suffix into a + deterministic but unique identifier string. + + Format: + --@ + Example: + AF-Q251019-9B3E2@001A + + Parameters + ---------- + domain : str + Two-to-six letter domain or organizational prefix (e.g. "AF"). + etype : str + One-to-three letter entity type identifier (e.g. "RQ" for Question). + estate : str + Lifecycle state whose one-letter code is appended to the version. + text : str + Entity-specific text seed (used to derive a stable hash). + version : int, optional + Starting version number (default is 1). + + Returns + ------- + FlexOID + A new Flex-O ID object with unique hash and digital signature. + + Notes + ----- + - Changing the text or state will produce a new hash and signature. + - Versions start at 1 and increment through `next_version()`. + """ + + if not (1 <= version <= FlexOID.MAX_VERSION): + raise ValueError(f"Version {version} exceeds limit; mark obsolete.") + + date_part = datetime.now(timezone.utc).strftime("%y%m%d") + + # include state in the seed so hash & signature depend on lifecycle state + seed = canonical_seed(f"{text}:{estate}") + + base_hash = FlexOID._blake_hash(seed) + unique_hash = FlexOID._ensure_unique(base_hash, version) + + ver_part = f"{version:03d}{estate}" + flexo_id_str = f"{domain}-{etype}{date_part}-{unique_hash}@{ver_part}" + + # short signature derived from state-aware seed + signature = hashlib.blake2s(seed.encode("utf-8"), digest_size=8).hexdigest().upper() + + return FlexOID(flexo_id_str, signature) + @property - def domain_id(self) -> str: - """I answer the domain prefix (e.g., 'PY_ARITH').""" - return self._m.group("domain_id") - - @property - def entity_type(self) -> str: - """I answer the short entity-type code (e.g., 'ITEM').""" - return self._m.group("entity_type") - - @property - def date_str(self) -> str: - """I answer the YYMMDD creation date as a string.""" - return self._m.group("date") - - @property - def date(self): - """I answer the creation date as a `datetime.date` instance.""" - return datetime.strptime(self.date_str, "%y%m%d").date() - - @property - def hash_part(self) -> str: - """I answer the twelve-hex BLAKE2s digest that stabilizes my prefix.""" - return self._m.group("hash") + def state_code(self): + part = self.flexo_id.rsplit("@", 1)[-1] + + if not (part and part[-1].isalpha()): + raise ValueError(f"Invalid Flex-O ID format: {self.flexo_id}") + return part[-1] @property def version(self) -> int: - """I answer my numeric version (1–999).""" - return int(self._m.group("version")) - - @property - def state_code(self) -> str: - """I answer my one-letter lifecycle state code (e.g., 'D', 'A', 'S').""" - return self._m.group("state") + try: + ver_state = self.flexo_id.rsplit("@", 1)[-1] + return int(ver_state[:-1]) # drop state suffix + except (ValueError, IndexError): + return 1 @property def prefix(self) -> str: - """I answer everything before the '@' symbol — my immutable lineage prefix.""" - return self.split('@', 1)[0] - - # ─────────────────────────────────────────── - # Transformations - # ─────────────────────────────────────────── - def with_state(self, new_state: str) -> "FlexOID": - """ - I create a copy of myself with the same version but a new state letter. - The given *new_state* must be a single capital letter. - """ - if not (isinstance(new_state, str) and len(new_state) == 1 and new_state.isalpha()): - raise ValueError("State must be a single capital letter.") - return FlexOID(f"{self.prefix}@{self.version:03d}{new_state}") - - def __repr__(self): - """I display myself in a developer-friendly representation.""" - return f"FlexOID({str(self)})" - - # ─────────────────────────────────────────── - # Generation helpers - # ─────────────────────────────────────────── - @staticmethod - def _blake_hash(text: str) -> str: - """ - I compute a 12-hex-digit BLAKE2s digest for *text*. - - I am modern, fast, and collision-resistant. - My output is deterministic and stable across platforms. - """ - - return hashlib.blake2s(text.encode("utf-8"), digest_size=6).hexdigest().upper() - - @staticmethod - def generate(domain: str, entity_type: str, state: str, text: str = "", version: int = 1): - """ - I create a new Flex-O ID. - - Identity is independent from content. - The hash part is now a cryptographic random nonce. - """ - - if not 1 <= version <= FlexOID.MAX_VERSION: - raise ValueError(f"Version {version} exceeds limit; mark obsolete.") - - if not (isinstance(state, str) and len(state) == 1 and state.isalpha()): - raise ValueError("state must be a single capital letter.") - - date_part = datetime.now(timezone.utc).strftime("%y%m%d") - - # 12 hex characters = 48-bit nonce (same visible size as before) - nonce = secrets.token_hex(6).upper() - - return FlexOID(f"{domain}-{entity_type}{date_part}-{nonce}@{version:03d}{state}") - - @staticmethod - def safe_generate(domain_id, entity_type, state, text="", version=1, repo=None): - """ - Generate a random identity and retry only if it already exists in repo. - """ - - if repo is None: - return FlexOID.generate(domain_id, entity_type, state, text="", version=version) - - while True: - oid = FlexOID.generate(domain_id, entity_type, state, text="", version=version) - existing = repo.get(str(oid)) if hasattr(repo, "get") else repo.get(oid) - if not existing: - return oid + return self.flexo_id.rsplit("@", 1)[0] + # ────────────────────────────────────────────────────────────────────────── @classmethod - def next_version(cls, oid: "FlexOID") -> "FlexOID": - """ - I create the next version within the same lineage. - I increment the numeric version but keep the same prefix and state. + def next_version(cls, oid) -> str: + """ + Create the next version in the same ID lineage. + + Increments the version counter of an existing FlexOID while preserving + its prefix and digital signature. Used when an entity transitions to a + new revision within the same lifecycle (e.g., minor updates or approvals). + + Parameters + ---------- + oid : FlexOID + The existing ID whose version is to be incremented. + + Returns + ------- + FlexOID + A new Flex-O ID with the same prefix and signature, but version +1. + + Raises + ------ + RuntimeError + If the maximum allowed version (`MAX_VERSION`) is exceeded. + + Notes + ----- + - The signature remains unchanged since the entity lineage is continuous. + - Warnings are logged when the version approaches obsolescence. """ new_ver = oid.version + 1 - if cls.WARN_THRESHOLD < new_ver < cls.MAX_VERSION: + + if new_ver > cls.WARN_THRESHOLD and new_ver < cls.MAX_VERSION: logger.warning(f"{oid} approaching obsolescence ({new_ver}/999).") if new_ver > cls.MAX_VERSION: raise RuntimeError(f"{oid} exceeded {cls.MAX_VERSION}; mark obsolete.") - return FlexOID(f"{oid.prefix}@{new_ver:03d}{oid.state_code}") + + new_id = f"{oid.prefix}@{new_ver:03d}{oid.state_code}" + return cls(new_id, oid.signature) + + # ────────────────────────────────────────────────────────────────────────── + @staticmethod + def clone_new_base(domain: str, etype: str, estate: str, text: str): + """ + Start a new Flex-O ID lineage for a derived or duplicated entity. + + This helper creates a completely new base ID (version 1) using the given + parameters, instead of incrementing an existing version chain. It is used + when an entity is copied, forked, or conceptually replaced by a new one. + + Returns + ------- + FlexOID + A new base ID starting at version 1, unrelated to the original lineage. + + Notes + ----- + - Equivalent to calling `generate(..., version=1)` explicitly. + - Used when creating "clones" or "variants" that should not share version history. + """ + return FlexOID.generate(domain, etype, estate, text, version=1) + + def __str__(self): + return self.flexo_id + + def __repr__(self): + return f"" @classmethod - def from_oid_and_version(cls, oid, version): - """ - I recreate *oid* with an explicitly given version number. - I keep its prefix and state, but replace the numeric counter. - """ - if not (1 <= version <= cls.MAX_VERSION): - raise ValueError(f"Version {version} out of bounds (1..{cls.MAX_VERSION}).") - return FlexOID(f"{oid.prefix}@{version:03d}{oid.state_code}") - - @staticmethod - def clone_new_base(domain_id: str, entity_type: str, state: str, text: str) -> "FlexOID": - """ - I start a completely new lineage (version 1) for a derived entity. - I am used when copying or forking an existing object that should - not share version history with its origin. - """ - return FlexOID.safe_generate(domain_id, entity_type, state, text, version=1) - - def to_dict(self) -> dict: - """ - I answer a dictionary that describes all my components — - useful for debugging, serialization tests, or human inspection. - """ - return { - "domain_id": self.domain_id, - "entity_type": self.entity_type, - "date": self.date, - "hash": self.hash_part, - "version": self.version, - "state": self.state_code, - } + def from_str(cls, id_str: str): + # reconstruct without a known signature + return cls(id_str, signature="") Index: exoentity/in_memory_backend.py =================================================================== --- flexoentity/in_memory_backend.py (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,41 +1,0 @@ -import json -from .persistance_backend import PersistanceBackend - - -class InMemoryBackend(PersistanceBackend): - """ - Persistence backend storing and returning dicts only (Option A). - """ - - def __init__(self, entity_class, storage=None): - super().__init__(entity_class=entity_class) - self._storage = storage if storage is not None else {} - - def save(self, entity_dict: dict): - flexo_id = entity_dict["meta"]["flexo_id"] - self._storage[flexo_id] = entity_dict - - def update(self, entity_dict: dict): - self.save(entity_dict) - - def delete(self, flexo_id: str): - self._storage.pop(flexo_id, None) - - def load(self, flexo_id: str): - return self._storage.get(flexo_id) - - def load_all(self): - return list(self._storage.values()) - - def clear(self): - self._storage.clear() - - # Optional file helpers still fine (dicts in/out) - def save_to_file(self, path): - with open(path, "w", encoding="utf-8") as f: - json.dump(list(self._storage.values()), f, ensure_ascii=False, indent=2) - - def load_from_file(self, path): - with open(path, "r", encoding="utf-8") as f: - data = json.load(f) - self._storage = {d["meta"]["flexo_id"]: d for d in data} Index: exoentity/json_file_backend.py =================================================================== --- flexoentity/json_file_backend.py (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,55 +1,0 @@ -import json -from .persistance_backend import PersistanceBackend -from .in_memory_backend import InMemoryBackend - - -class JsonFileBackend(PersistanceBackend): - """ - JSON-file based backend. - - Uses an internal InMemoryBackend and syncs to a single JSON file on disk. - Dict-only contract (Option A). - """ - - def __init__(self, entity_class, path): - super().__init__(entity_class=entity_class) - self._mem = InMemoryBackend(entity_class) - self._path = path - - # core API delegates (dicts) - - def save(self, entity_dict: dict): - self._mem.save(entity_dict) - - def update(self, entity_dict: dict): - self._mem.update(entity_dict) - - def delete(self, flexo_id: str): - self._mem.delete(flexo_id) - - def load(self, flexo_id: str): - return self._mem.load(flexo_id) - - def load_all(self): - return self._mem.load_all() - - def clear(self): - self._mem.clear() - - # file sync (dicts) - - def flush_to_file(self): - data = self._mem.load_all() - with open(self._path, "w", encoding="utf-8") as f: - json.dump(data, f, ensure_ascii=False, indent=2) - - def load_from_file(self): - try: - with open(self._path, "r", encoding="utf-8") as f: - data = json.load(f) - except FileNotFoundError: - return - - self._mem.clear() - for d in data: - self._mem.save(d) Index: exoentity/logger.py =================================================================== --- flexoentity/logger.py (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,24 +1,0 @@ -import logging -import logging.config -import os -import tempfile -from pathlib import Path - - -log_dir = Path(os.environ.get('FLEXO_LOG_DIR', tempfile.gettempdir())) - -log_path = log_dir / 'flexo.log' - -if not log_path.exists(): - try: - log_path.touch() - except PermissionError: - conf_file = Path(__file__).parent / 'flexo_logging.conf' - print("Conf file", conf_file) - logging.config.fileConfig(conf_file) -else: - logging.basicConfig(filename=str(log_path), encoding='utf-8', level=logging.DEBUG) - -logger = logging.getLogger(__name__) - -logger.setLevel(os.environ.get('FLEXO_LOG_LEVEL', logger.getEffectiveLevel())) Index: exoentity/persistance_backend.py =================================================================== --- flexoentity/persistance_backend.py (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,52 +1,0 @@ -class PersistanceBackend: - """ - Interface for all persistence backends. - - Contract (Option A): - - Backends store serialized entity dictionaries. - - Backends return dictionaries (never FlexoEntity instances). - - EntityManager is responsible for dict <-> entity conversion. - - Stored dict format must include: - - meta.flexo_id (string) - - meta.schema.name / meta.schema.version - - content (entity-specific) - """ - - def __init__(self, entity_class): - self._entity_class = entity_class - - @property - def entity_class(self): - return self._entity_class - - @entity_class.setter - def entity_class(self, a_class): - self._entity_class = a_class - - # ------------------------- - # Write API (dict) - # ------------------------- - - def save(self, entity_dict: dict) -> None: - raise NotImplementedError - - def update(self, entity_dict: dict) -> None: - raise NotImplementedError - - def delete(self, flexo_id: str) -> None: - raise NotImplementedError - - # ------------------------- - # Read API (dict) - # ------------------------- - - def load(self, flexo_id: str) -> dict | None: - raise NotImplementedError - - def load_all(self) -> list[dict]: - raise NotImplementedError - - def clear(self): - """Remove all stored entities.""" - raise NotImplementedError Index: exoentity/runtime_backend.py =================================================================== --- flexoentity/runtime_backend.py (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,32 +1,0 @@ -from .persistance_backend import PersistanceBackend - - -class RuntimeBackend(PersistanceBackend): - """ - Runtime backend (Option A). - - - Stores dicts only (no entity instances) - - Useful as an in-process cache - """ - - def __init__(self, entity_class): - super().__init__(entity_class=entity_class) - self._store: dict[str, dict] = {} - - def save(self, entity_dict: dict): - self._store[entity_dict["meta"]["flexo_id"]] = entity_dict - - def update(self, entity_dict: dict): - self.save(entity_dict) - - def delete(self, flexo_id: str): - self._store.pop(flexo_id, None) - - def load(self, flexo_id: str): - return self._store.get(flexo_id) - - def load_all(self): - return list(self._store.values()) - - def clear(self): - self._store.clear() Index: exoentity/signing_backends.py =================================================================== --- flexoentity/signing_backends.py (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,214 +1,0 @@ -import subprocess -import platform -from pathlib import Path -import tempfile -import uuid -from abc import ABC, abstractmethod - -class SigningBackend(ABC): - def __init__(self, cert_ref): - """ - cert_ref is an object that is expected to provide (depending on platform): - - Linux: - - identifier: path to certificate (PEM) - - private_key_path: path to private key (PEM) - - public_cert_path: optional, defaults to identifier if missing - - Windows: - - identifier: certificate thumbprint (SHA1; with or without colons) - - macOS: - - identifier: certificate Common Name (CN) - - public_cert_path: path to public certificate (PEM) - """ - self.cert_ref = cert_ref - - @property - @abstractmethod - def certificate_thumbprint(self) -> str: - """Return canonical thumbprint for Signature entity storage.""" - raise NotImplementedError - - @abstractmethod - def sign(self, data: bytes) -> bytes: - """Return PKCS#7 DER signature bytes.""" - raise NotImplementedError - - @abstractmethod - def verify(self, data: bytes, signature: bytes) -> bool: - """Return True/False after verifying PKCS#7 signature.""" - raise NotImplementedError - - def _tmp(self, suffix: str) -> Path: - """Return a unique temporary file path for this process.""" - return Path(tempfile.NamedTemporaryFile(delete=False).name).with_suffix(suffix) - - -class LinuxOpenSSLBackend(SigningBackend): - - @property - def certificate_thumbprint(self) -> str: - out = subprocess.check_output([ - "openssl", "x509", "-in", self.cert_ref.identifier, - "-noout", "-fingerprint", "-sha1" - ]).decode() - return out.strip().split("=")[1].replace(":", "") - - def sign(self, data: bytes) -> bytes: - datafile = self._tmp(".bin") - sigfile = self._tmp(".p7s") - datafile.write_bytes(data) - - subprocess.check_call([ - "openssl", "cms", "-sign", - "-binary", - "-in", str(datafile), - "-signer", self.cert_ref.identifier, - "-inkey", self.cert_ref.private_key_path, # operator-provided - "-outform", "DER", - "-out", str(sigfile) - ]) - - out = sigfile.read_bytes() - datafile.unlink() - sigfile.unlink() - return out - - def verify(self, data: bytes, signature: bytes) -> bool: - datafile = self._tmp(".bin") - sigfile = self._tmp(".p7s") - datafile.write_bytes(data) - sigfile.write_bytes(signature) - - try: - subprocess.check_call([ - "openssl", "cms", "-verify", - "-in", str(sigfile), - "-inform", "DER", - "-content", str(datafile), - "-CAfile", self.cert_ref.public_cert_path, - "-purpose", "any", - "-out", "/dev/null", - ]) - return True - except subprocess.CalledProcessError: - return False - finally: - datafile.unlink() - sigfile.unlink() - -class WindowsCertutilBackend(SigningBackend): - - @property - def certificate_thumbprint(self) -> str: - return self.cert_ref.identifier.upper().replace(":", "") - - def sign(self, data: bytes) -> bytes: - datafile = self._tmp(".bin") - sigfile = self._tmp(".p7s") - datafile.write_bytes(data) - - cmd = [ - "certutil", - "-user", - "-silent", - "-p", "", - "-sign", - str(datafile), - str(sigfile), - self.certificate_thumbprint - ] - - subprocess.check_call(cmd) - out = sigfile.read_bytes() - datafile.unlink() - sigfile.unlink() - return out - - def verify(self, data: bytes, signature: bytes) -> bool: - datafile = self._tmp(".bin") - sigfile = self._tmp(".p7s") - datafile.write_bytes(data) - sigfile.write_bytes(signature) - - try: - subprocess.check_call([ - "certutil", "-verify", str(sigfile), str(datafile) - ]) - return True - except subprocess.CalledProcessError: - return False - finally: - datafile.unlink() - sigfile.unlink() - -class MacOSSecurityCMSBackend(SigningBackend): - - @property - def certificate_thumbprint(self) -> str: - # Derive from provided public cert - out = subprocess.check_output([ - "openssl", "x509", "-in", self.cert_ref.public_cert_path, - "-noout", "-fingerprint", "-sha1" - ]).decode() - return out.strip().split("=")[1].replace(":", "") - - def sign(self, data: bytes) -> bytes: - datafile = self._tmp(".bin") - sigfile = self._tmp(".p7s") - datafile.write_bytes(data) - - subprocess.check_call([ - "security", "cms", "-S", - "-N", self.cert_ref.identifier, # CN - "-i", str(datafile), - "-o", str(sigfile) - ]) - - out = sigfile.read_bytes() - datafile.unlink() - sigfile.unlink() - return out - - def verify(self, data: bytes, signature: bytes) -> bool: - sigfile = self._tmp(".p7s") - datafile = self._tmp(".bin") - sigfile.write_bytes(signature) - datafile.write_bytes(data) - - # explicit verify (preferred) - if self.cert_ref.public_cert_path: - try: - subprocess.check_call([ - "openssl", "cms", "-verify", - "-in", str(sigfile), - "-inform", "DER", - "-content", str(datafile), - "-CAfile", self.cert_ref.public_cert_path, - "-purpose", "any", - "-out", "/dev/null", - ]) - return True - except subprocess.CalledProcessError: - return False - - # fallback: structural validation only - try: - subprocess.check_call(["security", "cms", "-D", "-i", str(sigfile)]) - return True - except subprocess.CalledProcessError: - return False - finally: - sigfile.unlink() - datafile.unlink() - -def get_signing_backend(cert_ref): - system = platform.system() - if system == "Linux": - return LinuxOpenSSLBackend(cert_ref) - if system == "Windows": - return WindowsCertutilBackend(cert_ref) - if system == "Darwin": - return MacOSSecurityCMSBackend(cert_ref) - raise RuntimeError("Unsupported platform") Index: exoentity/sqlite_entity_backend.py =================================================================== --- flexoentity/sqlite_entity_backend.py (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,114 +1,0 @@ -import json -from .persistance_backend import PersistanceBackend - - -class SQLiteEntityBackend(PersistanceBackend): - """ - SQLite backend storing and returning **dicts only**. - - Note: - - EntityManager performs dict <-> entity conversion. - - This backend does not call entity_class.from_dict(). - """ - - def __init__(self, entity_class, conn, table_name): - super().__init__(entity_class) - self.conn = conn - self.table = table_name - self._init_schema() - - def _init_schema(self): - self.conn.execute(f""" - CREATE TABLE IF NOT EXISTS {self.table} ( - flexo_id TEXT PRIMARY KEY, - json TEXT NOT NULL - ) - """) - self.conn.commit() - - # ------------------------- - # Helpers - # ------------------------- - - def _ensure_dict(self, entity_or_dict) -> dict: - """ - Transitional helper: - - Accept dict (preferred) - - Accept FlexoEntity-like objects with .to_dict() (legacy) - """ - if isinstance(entity_or_dict, dict): - return entity_or_dict - if hasattr(entity_or_dict, "to_dict"): - return entity_or_dict.to_dict() - raise TypeError(f"Expected dict or object with to_dict(), got {type(entity_or_dict).__name__}") - - def _flexo_id_from_dict(self, entity_dict: dict) -> str: - try: - return entity_dict["meta"]["flexo_id"] - except Exception as e: - raise KeyError("entity_dict must contain meta.flexo_id") from e - - # ------------------------- - # Writes (dict) - # ------------------------- - - def save(self, entity_dict): - d = self._ensure_dict(entity_dict) - fid = self._flexo_id_from_dict(d) - self.conn.execute( - f"INSERT OR REPLACE INTO {self.table} (flexo_id, json) VALUES (?, ?)", - (fid, json.dumps(d)) - ) - self.conn.commit() - - def update(self, entity_dict): - d = self._ensure_dict(entity_dict) - fid = self._flexo_id_from_dict(d) - self.conn.execute( - f"UPDATE {self.table} SET json = ? WHERE flexo_id = ?", - (json.dumps(d), fid) - ) - self.conn.commit() - - def delete(self, flexo_id): - self.conn.execute( - f"DELETE FROM {self.table} WHERE flexo_id = ?", - (flexo_id,) - ) - self.conn.commit() - - # ------------------------- - # Reads (dict) - # ------------------------- - - def load(self, flexo_id): - row = self.conn.execute( - f"SELECT json FROM {self.table} WHERE flexo_id = ?", - (flexo_id,), - ).fetchone() - - if not row: - return None - - # row indexing depends on connection row_factory; support both - raw = row["json"] if isinstance(row, dict) or hasattr(row, "__getitem__") else row[0] - if not isinstance(raw, str): - raw = raw[0] - return json.loads(raw) - - def load_all(self): - rows = self.conn.execute( - f"SELECT json FROM {self.table}" - ).fetchall() - - result = [] - for r in rows: - raw = r["json"] if isinstance(r, dict) or hasattr(r, "__getitem__") else r[0] - if not isinstance(raw, str): - raw = raw[0] - result.append(json.loads(raw)) - return result - - def clear(self): - self.conn.execute(f"DELETE FROM {self.table}") - self.conn.commit() Index: exoentity/typed_collection.py =================================================================== --- flexoentity/typed_collection.py (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,44 +1,0 @@ -from flexoentity.flexo_collection import FlexoCollection -from flexoentity import FlexoEntity - -class TypedCollection(FlexoCollection): - """ - A type-safe collection of FlexOEntity subclasses. - """ - - def __init__(self, entity_class, items=None): - if not issubclass(entity_class, FlexoEntity): - raise TypeError("entity_class must be a subclass of FlexOEntity") - - self.entity_class = entity_class - super().__init__() - - if items: - for e in items: - self.add(e) - - def add(self, entity: FlexoEntity, key=None): - if not isinstance(entity, self.entity_class): - raise TypeError( - f"TypedCollection[{self.entity_class.__name__}] " - f"cannot accept {type(entity).__name__}" - ) - super().add(entity, key) - - # ------------------------------------------------------------------ - # Deserialization - # ------------------------------------------------------------------ - @classmethod - def deserialize(cls, entity_class, list_of_dicts): - """ - Create a TypedCollection of `entity_class` from a list of dicts. - """ - # create empty collection - coll = cls(entity_class) - - # build entity objects - for d in list_of_dicts: - entity = entity_class.from_dict(d) - coll.add(entity) - - return coll Index: g/FlexoEntityTalk.org =================================================================== --- org/FlexoEntityTalk.org (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,224 +1,0 @@ -* Tech-Talk - Hashes / IDs / Lebenszyklen / FlexoEntity -:PROPERTIES: -:CUSTOM_ID: hashes-ids-lifecycle -:END: - -** Was ist ein Hash? - -Ein Hash ist ein hoffentlich eindeutiger "Fingerabdruck" für Daten, der -sich aus eben diesen Daten über den Algorithmus einer Hash-Funktion errechnet. - -Die Eingabedaten bezeichnet man als Seed, das Ergebnis als Digest. - -Man gibt die Saat (das Futter) an die Hashfunktion zum Zerkauen/Zerkleinern (Hashen), -das wird verdaut, nebenbei noch gewürzt (Salt) und kommt als Digest wieder heraus. -Die Länge der Ausgabe ist fest. - -** Ein wichtiger Punkt - -Wichtig ist auch zu verstehen, dass Hashes sogenannte "One-Way-Funktionen" sind. -Das heißt, man kann das Hashing nicht einfach umkehren, um das ursprüngliche Objekt wiederherzustellen. -Ein Hash ist also eine Einbahnstraße: Er dient nur dazu, die Daten zu repräsentieren, -aber man kann nicht vom Hash auf die originalen Daten zurückrechnen. -Aus Kacke kann man kein Essen zaubern. - -** Was ist kein Hash? - -Die ISBN identifiziert ein Buch eindeutig. Sie ist nicht das Buch selbst, sondern eher ein -Identifier (eine geordnete Nummer mit Prüfziffer). Sie ist aber kein Hash, weil sie nicht -aus dem Inhalt des Buches berechnet wird. Ähnliches gilt für die IBAN - -|-------------------------+---------------------------+-------------------------------------| -| Merkmal | Hash | Identifier (z. B. ISBN, IBAN, UUID) | -|-------------------------+---------------------------+-------------------------------------| -| Abhängig vom Inhalt | ja | nein | -| Ziel | Integrität / Vergleich | Identität / Referenz | -| Länge | fix z. B. 32 Hex-Zeichen | frei definierbar | -| Zufällig oder berechnet | deterministisch berechnet | oft generiert oder vergeben | -|-------------------------+---------------------------+-------------------------------------| - -** Warum verwenden wir Hashes? - -Hashes sind super praktisch, weil sie uns eine Menge Zeit und Rechenleistung sparen. -Anstatt zwei große Dateien oder ganze Bücher direkt miteinander zu vergleichen, -vergleichen wir einfach ihre Hashes. Wenn die Hashes gleich sind, wissen wir, -dass die Daten sehr wahrscheinlich identisch sind. Wenn sie unterschiedlich sind, -hat sich etwas geändert. Das spart enorm viel Aufwand. - -** Beispiele für den Einsatz von Hashes - -- **Passwort-Hashes**: Wenn du ein Passwort eingibst, wird oft nur der Hash gespeichert, nicht das - eigentliche Passwort. So bleibt es sicher, selbst wenn jemand die Datenbank stiehlt. - - Beispiel aus der /etc/shadow-Datei - -root:$6$tsicHaoV3Q$YtAbiIvrHGXFtAJYz9tcEYWHXiGVQ40sJAgzPAbc57lIq9jH8eYjWXwctSW6YQnrMznRFcm6yXLnnY9mHhso20 -enno:$6$sdygzfEgx0$YpaZJMQdkZgxGPclphz6RojqNG.PSNEq1oIHRP4kvZRN2iuS5MQrxt0nCkrYQIcpDGyohrb1o0S/GkWrFriWL1 - -Der Hash ist länger als das Passwort. Das dient der Sicherheit, weil man zwar nicht zurückrechnen, -wohl aber vorwärtsrechnen kann. Hash-Funktionen sind deterministisch. -Bei Linux-Passworteinträgen hat man hier $6$ den Hashalgorithmus kodiert, im daraufolgendem $.....$ das Salz -und im Rest den eigentlichen Hash kodiert - -- **Git-Commits**: In Versionskontrollsystemen wie Git werden Hashes benutzt, um jeden Schwung an - Änderungen mit einer eindeutigen Kennung zu versehen - -- **URL-Shortener: hash("https://example.com") → 8a3f12 - -** FlexOID - -Die FlexOID ist eine Mischform. Sie selbst ist ein Identifier, der als Bestandteil aber einen Hash -enthält, um möglichst eindeutig, aber nicht zu lang zu sein. - -AF-I251022-70F759@001D -│ │ │ │ │ │ -│ │ │ │ │ └── State (Draft, Approved, Signed, Published, Obsolete) -│ │ │ │ └──── Version -│ │ │ └────────── Hash (6 bytes, 12 Stellen) -│ │ └────────────────── Date (YYMMDD) -│ └──────────────────── Entity type (ITEM) -└─────────────────────── Domain (e.g. Air force) - -In der ersten Variante der FlexOID habe ich mich mit einem 6-stelligen Hash zufrieden gegeben, -weil das einfach lesbarer ist. Warum reicht das nicht? - -** Das Geburtstags-Paradoxon - -Der obige 3-Byte Hash liefert mir etwas über 16 Millionen unterschiedlich Varianten. -Das klingt viel. Ist es aber nicht. Wieviele Schüler müssen nacheinander die Klasse -betreten bis sich zwei finden, die mit mehr als 50 Prozent Wahrscheinlichkeit am gleichen -Tag Geburtstag (also ein identisches Merkmal) haben? - -** Lösung - -Ab 23 Schülern beträgt die Wahrscheinlichkeit 50 % (näherungsweise Wurzel 365 Tagen) - -Bei 47 Schülern beträgt die Wahrscheinlichkeit bereits 95 Prozent - -Unsere 3 Bytes ergeben zwar 16.000.000 mögliche Varianten - im Gegenzug zu den 365 Tagen im Jahr -aus dem Geburtstagsbeispiel - aber da die Wahrscheinlichkeit zur Kollision hier bei -etwa Wurzel 16 Mio. liegt, bekommt man bereits bei 4000 Neuzugängen die ersten -Übereinstimmungen im Hash. - -Der Hash ist also nicht gut genug, weil man sehr schnell und sehr häufig, diese -Übereinstimmungen feststellen und behandeln müsste, wenn man weiterhin eindeutige -Zuordnungen treffen will. - -Wenn man die Ausgabe der Hashfunktion auf 6 Bytes erweitert, kommen die ersten Kollisionen -erst bei etwa 20 Mio erzeugten Hashes (Fingerabdrücken) und die kann man dann ohne -Einbußen gesondert behandeln (Salzen), weil es so selten passiert. - -Übrigens, wenn man beim Menschen den Daumenabdruck nimmt und sich dabei auf 12 Merkmale -beschränkt und sehr lockere Toleranzen ansetzt (was man in der Praxis nicht macht), -hat man bereits nach 14000 Menschen eine Übereinstimmung. Bei 24 Merkmalen und sehr -lockeren Toleranzen, hat man bei etwa nach 9 Mio. Menschen eine ungefähre Übereinstimmung. -Da muss die Polizei schon sehr schlampig arbeiten, damit man fälschlicherweise beschuldigt wird. -Die Zahlen in der Realität sind sogar noch deutlich höher. - -** FlexoEntity - -Nun haben wir gesehen, dass wir mit der FlexOID (mit 6-Byte Hash) sehr viele unterschiedliche -Dinge eindeutig bestimmen können. Da unsere FlexOID erstmal nur eine Zeichenfolge ist, -brauchen wir etwas das damit umgehen kann und was dafür verantwortlich ist. Das ist die FlexoEntity. - -Sie beinhaltet zusätzlich ein Origin-Feld, wo festgehalten wird, woher diese -Entität stammt (beispielsweise aus einer Hashkollision oder einer Änderung an den weiteren Daten) - -Jede Klasse, die von FlexoEntity erbt, muss zwingend die Methode "text_seed" implementieren, -mit der der Algorithmus einer Hash-Funktion gefüttert wird, aus der dann der 6-Byte Hash herauspurzelt. -Hashfunktionen sind z.B. MD5, SHA1, SHA256 oder wie von mir genutzt: Blake2s. -Die Mathematik dahinter ist recht aufwändig, aber wer sich mal einlesen möchte - -- Hashing in Smalltalk: Theory and Practice von Andres Valloud - -Damit die Hash-Funktion genug Eingabedaten pro Entity hat, muss man sich überlegen, welche Merkmale -einer Entität man durch "text_seed" übermittelt. - -** text_seed - -Man kann beliebige Klassen von FlexoEntity ableiten und erhält ohne Aufwand die Funktionalität -zur eindeutigen Identifizierung und zur Lebenszyklus-Verwaltung - -Das ist das Beispiel einer ChoiceQuestion, also einer Testfrage, wo mögliche Antworten enthalten sind - - @property - def text_seed(self) -> str: - """Include answer options (and points) for deterministic ID generation.""" - base = super().text_seed - if not self.options: - return base - - joined = "|".join( - f"{opt.text.strip()}:{opt.points}" - for opt in sorted(self.options, key=lambda o: o.text.strip().lower()) - ) - return f"{base}::{joined}" - -** Lebenszyklus - -Der Lebenszyklus einer Entität folgt dieser Reihenfolge und ist nicht umkehrbar - -- Entwurf (DRAFT) -- Genehmigt (APPROVED) -- Unterschrieben (APPROVED_AND_SIGNED) -- Veröffentlicht (PUBLISHED) -- Veraltet (OBSOLETE) - -Eine Entität, die bereits die Stufe Veröffentlicht erreicht hat, kann nicht in die Stufe -(nur) Unterschrieben zurückkehren. Daher ist auch die Lebenszyklusstufe in der ID kodiert -(letztes Symbol der FlexOID) - -Beispiele für Entitäten: - -- Testfrage -- Fragenkatalog -- Einstufungstest -- Zertifikat - -Ein veröffentlichter Einstufungstest kann nur Fragen beinhalten, die ihrerseits die Stufe Veröffentlicht -erreicht haben. Ein Zertifikat kann nur ausgestellt werden, wenn der passende Einstufungstest die Stufe -Veröffentlicht hat. Das origin-Feld des Zertifikats sollte sinnvollerweise die ID des Tests enthalten. - -** Erhöhung der Versionsnummer oder neue ID - -Sobald - aus Gründen - eine neue ID vergeben werden muss, wird ggf. die Ursprungs-ID -im Feld origin der neuen FlexoEntity gespeichert. So ist immer ein Suchen im Stammbaum möglich. - -AF-Q251022-70F759@001D -│ │ │ │ │ │ -│ │ │ │ │ └── State (Draft, Approved, Signed, Published, Obsolete) -│ │ │ │ └──── Version -│ │ │ └────────── Hash (3 bytes, 6 Stellen) -│ │ └────────────────── Date (YYMMDD) -│ └──────────────────── Entity type (Question) -└─────────────────────── Domain (e.g. Air force) - - -Eine einfache Erhöhung der Versionsnummer (am Ende der ID) ist unter Umständen auch ausreichend, -damit nicht zu häufig komplett neue IDs erzeugt werden müssen. Der Entscheidungsbaum dazu ist -für das Projekt FlexoGrader recht umfangreich und soll hier nicht weiter besprochen werden. -Da es aber im Hintergrund passiert und vom Endanwender nicht bemerkt wird, behindert der -Mechanismus die Nutzung des FlexoGrader nicht. - -** Reifegrad - -Das Konzept und Design hat die Alpha-Phase verlassen, der Code ist aber noch deutlich Beta. -RC1 etwa Anfang Dezember (nach meinem Urlaub). Änderungen an der FlexOID sind nicht mehr zu erwarten, beim API der FlexoEntity schon eher. - -Das Perfekte ist der Feind des Guten, aber ... - -Als Gegenbeispiel für unausgereiftes Design ist Pythons eingebaute datetime Bibliothek. -Da könnte man ein Buch drüber schreiben. Halb Drama, halb Komödie. - -- dateutil, arrow, pendulum, maya, moment, delorean, pytz, zoneinfo + numpy/pandas - -** Lizenz - -Da ich die FlexoEntity-Bibliothek in meiner Freizeit entworfen und programmiert habe, -bin ich alleiniger Urheber. Als Lizenz habe ich MIT gewählt, somit kann - unter Nennung -des Urhebers - jeder damit machen, was er möchte - -Der FlexoGrader ist ein Bundeswehrprojekt, welches ich im Dienst programmiere und somit -gehen die Nutzungsrechte uneingeschränkt auf den Dienstherren über. - -Das betrifft auch etwaige Folgeprojekte wie FlexoVault und FlexoDrill Index: g/README.org =================================================================== --- org/README.org (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,294 +1,0 @@ -#+TITLE: flexoentity -#+SUBTITLE: A hardened entity base and deterministic identifier system for the Flex-O family -#+AUTHOR: Enno -#+DATE: 2025-10-20 -#+OPTIONS: toc:3 num:nil - -* Overview - -`flexoentity` provides the *identity and lifecycle backbone* for all Flex-O components -(Flex-O-Grader, Flex-O-Vault, Flex-O-Drill, …). - -It defines how entities such as questions, media, catalogs, and exams are *identified, versioned, signed, and verified* — all without any external database dependencies. - -At its heart lie two modules: - -- =id_factory.py= – robust, cryptographically-verifiable *Flex-O ID generator* -- =flexo_entity.py= – abstract *base class for all versioned entities* - -Together, they form a compact yet powerful foundation for audit-ready, reproducible data structures across offline and air-gapped deployments. - -- Design Goals - -|----------------|--------------------------------------------------------------------------------------------------------| -| Goal | Description | -|----------------|--------------------------------------------------------------------------------------------------------| -| *Determinism* | IDs are derived from canonicalized entity content — identical input always yields identical ID prefix. | -| *Integrity* | BLAKE2s hashing and digital signatures protect against manual tampering. | -| *Traceability* | Version numbers (=@001A=, =@002S=, …) track entity lifecycle transitions. | -| *Stability* | Hash prefixes remain constant across state changes; only version and state suffixes evolve. | -| *Auditability* | Every entity can be serialized, verified, and reconstructed without hidden dependencies. | -| *Simplicity* | Pure-Python, zero external libraries, self-contained and easy to embed. | -|----------------|--------------------------------------------------------------------------------------------------------| - -* Flex-O ID Structure - -Each entity carries a unique *Flex-O ID*, generated by =FlexOID.generate()=. - -#+BEGIN_EXAMPLE -AF-I250101-9A4C2D@003S -#+END_EXAMPLE - -|-----------+-----------------+----------------------------------------------| -| Segment | Example | Meaning | -|-----------+-----------------+----------------------------------------------| -| *Domain* | =AF or PY_LANG= | Uppercase - Logical scope (e.g. "Air Force") | -| *Type* | =I= | Entity type (e.g. ITEM) | -| *Date* | =250101= | UTC creation date (YYMMDD) | -| *Hash* | =9A4C2D4F6E53= | 12-digit BLAKE2s digest of canonical content | -| *Version* | =003= | Sequential version counter | -| *State* | =S= | Lifecycle state (D, A, S, P, O) | -|-----------+-----------------+----------------------------------------------| - - -* Lifecycle States - -|-----------------------|---------|-----------------------------| -| State | Abbrev. | Description | -|-----------------------|---------|-----------------------------| -| *DRAFT* | =D= | Editable, not yet validated | -| *APPROVED* | =A= | Reviewed and accepted | -| *APPROVED_AND_SIGNED* | =S= | Cryptographically signed | -| *PUBLISHED* | =P= | Released to consumers | -| *OBSOLETE* | =O= | Archived or replaced | -|-----------------------|---------|-----------------------------| - -Transitions follow a strict progression: -#+BEGIN_EXAMPLE -DRAFT -> APPROVED -> APPROVED_AND_SIGNED -> PUBLISHED -> OBSOLETE -#+END_EXAMPLE - -Only DRAFT entities can be deleted - all others got OBSOLETE mark instead - -* Core Classes - -** FlexOID - -A lightweight immutable class representing the full identity of an entity. - -*Highlights* -- safe_generate(domain, entity_type, estate, text, version=1, repo) -> create a new ID -- next_version(oid) -> increment version safely -- clone_new_base(domain, entity_type, estate, text) -> start a new lineage -- Deterministic prefix, state-dependent signature - -** =FlexoEntity= -Abstract base class for all versioned entities (e.g., Question, Exam, Catalog). - -Implements: -- ID lifecycle management (approve(), sign(), publish(), obsolete()) -- Serialization (to_json(), from_json(), to_dict(), from_dict()) -- Integrity verification (verify_integrity(entity)) -- Controlled state transitions with automatic timestamps - -Subclasses define a single property: - -#+BEGIN_SRC python -@property -def text_seed(self) -> str: - """Canonical text or core content for hashing.""" -#+END_SRC - -* Integrity Verification - -Each entity can self-verify its integrity: - -#+BEGIN_SRC python -entity = Question.with_domain_id(domain_id="AF", text="What is Ohm’s law?", topic="Electronics") -json_str = entity.to_json() -reloaded = Question.from_json(json_str) - -assert FlexoEntity.verify_integrity(reloaded) -#+END_SRC - -If the file is tampered with (e.g. "Ohm’s" → "Omm’s"), verification fails: - -* Real World Example - -Below you can see the implementation of a dedicated FlexoEntity class, used for Domains. -We set an ENTITY_TYPE and define the needed fields in the data class. We define how to create -a default object, the text_seed (it is easy because the domain id is unique and therefore sufficient) -and the methods for serialization. - -#+BEGIN_SRC python -from uuid import UUID -from dataclasses import dataclass -from flexoentity import FlexOID, FlexoEntity, EntityType - -@dataclass -class Domain(FlexoEntity): - """ - I am a helper class to provide more information than just a - domain abbreviation in FlexOID, doing mapping and management - """ - - ENTITY_TYPE = EntityType.DOMAIN - - fullname: str = "" - description: str = "" - classification: str = "UNCLASSIFIED" - - @classmethod - def default(cls): - """Return the default domain object.""" - return cls.with_domain_id(domain_id="GEN_GENERIC", - fullname="Generic Domain", classification="UNCLASSIFIED") - - @property - def text_seed(self) -> str: - return self.domain_id - - def to_dict(self): - base = super().to_dict() - base.update({ - "flexo_id": self.flexo_id, - "domain_id": self.domain_id, - "fullname": self.fullname, - "description": self.description, - "classification": self.classification, - }) - return base - - @classmethod - def from_dict(cls, data): - # Must have flexo_id - if "flexo_id" not in data: - raise ValueError("Domain serialization missing 'flexo_id'.") - - flexo_id = FlexOID(data["flexo_id"]) - - obj = cls( - fullname=data.get("fullname", ""), - description=data.get("description", ""), - classification=data.get("classification", "UNCLASSIFIED"), - flexo_id=flexo_id, - _in_factory=True - ) - - # Restore metadata - obj.origin = data.get("origin") - obj.fingerprint = data.get("fingerprint", "") - obj.originator_id = ( - UUID(data["originator_id"]) if data.get("originator_id") else None - ) - obj.owner_id = ( - UUID(data["owner_id"]) if data.get("owner_id") else None - ) - - return obj -#+END_SRC - -* Usage -#+BEGIN_SRC python -d = Domain.default() -print(d.flexo_id) # GEN_GENERIC-D251124-67C2CAE292CE@001D -d.approve() -print(d.flexo_id) # GEN_GENERIC-D251124-67C2CAE292CE@001A -d.sign() -print(d.flexo_id) # GEN_GENERIC-D251124-67C2CAE292CE@001S -#+END_SRC - -* Serialization Example - -to_dict() - -#+BEGIN_SRC python -{ - 'flexo_id': FlexOID(GEN_GENERIC-D251124-29CE0F4BE59D@001S), - 'fingerprint': '534BD2EC5C5511F1', - 'origin': FlexOID(GEN_GENERIC-D251124-67C2CAE292CE@001D), - 'originator_id': '00000000-0000-0000-0000-000000000000', - 'owner_id': '00000000-0000-0000-0000-000000000000', - 'domain_id': 'GEN_GENERIC', - 'fullname': 'Generic Domain', - 'description': '', - 'classification': 'UNCLASSIFIED'} -#+END_SRC - -to_json() - -#+BEGIN_SRC js -{ - "flexo_id": "GEN_GENERIC-D251124-29CE0F4BE59D@001S", - "fingerprint": "534BD2EC5C5511F1", - "origin": "GEN_GENERIC-D251124-67C2CAE292CE@001D", - "originator_id": "00000000-0000-0000-0000-000000000000", - "owner_id": "00000000-0000-0000-0000-000000000000", - "domain_id": "GEN_GENERIC", - "fullname": "Generic Domain", - "description": "", - "classification": "UNCLASSIFIED" -} -#+END_SRC - - -* Entity Type and State Codes - -|-------------+------+----------------------------------------------------------------------------| -| EntityType | Code | Typical Use | -|-------------+------+----------------------------------------------------------------------------| -| GENERIC | G | Generic entities that does not fit other types yet or are temporarily only | -| DOMAIN | D | Every Domain is of this type | -| MEDIA | M | Every media item belongs to this type, e.g. Pictures, Audio, Video | -| ITEM | I | An Entity what is usually used in a collection, e.g. Questions in a test | -| COLLECTION | C | A collection of items, as an Exam or a catalog | -| TEXT | T | A text document | -| HANDOUT | H | A published document | -| OUTPUT | O | The output of a computation | -| RECORD | R | Record type data, as bibliography entries | -| SESSION | S | A unique session, e.g. managed by a session manager -| USER | U | User objects | -| CONFIG | F | CONFIG files that need to be tracked over time and state | -| EVENT | E | Events that have to be tracked over time, as status messages or orders | -| ATTESTATION | X | Entities that attest a formal technical (not human) check e.g. Signatures | -|-------------+------+----------------------------------------------------------------------------| - -|---------------------|------|-------------------| -| EntityState | Code | Meaning | -|---------------------|------|-------------------| -| DRAFT | D | Work in progress | -| APPROVED | A | Reviewed | -| APPROVED_AND_SIGNED | S | Signed version | -| PUBLISHED | P | Publicly released | -| OBSOLETE | O | Deprecated | -|---------------------|------|-------------------| - -* Design Notes -- *Hash Stability:* Only domain, entity type, and content text influence the hash. - This ensures consistent prefixes across state changes. -- *State-Dependent Signatures:* Each lifecycle stage has its own signature seed. - Modifying a file without re-signing invalidates integrity. -- *Obsolescence Threshold:* Version numbers above 900 trigger warnings; - beyond 999 are considered obsolete. -- *Clone Lineages:* Cloning an entity resets versioning but preserves metadata lineage. - -* Dependencies -- Python 3.11+ -- Standard library only (=hashlib=, =json=, =datetime=, =enum=, =dataclasses=) - -No external packages. Fully compatible with *Guix*, *air-gapped* deployments, and *reproducible builds*. - -* Integration -`flexoentity` is imported by higher-level modules such as: - -- *Flex-O-Grader* → manages question catalogs and exam bundles -- *Flex-O-Vault* → provides persistent media storage with metadata integrity -- *Flex-O-Drill* → uses versioned entities for training simulations - -All share the same identity and versioning logic — ensuring that -*what was approved, signed, and published remains provably authentic.* - -* License -MIT License 2025 -Part of the *Flex-O family* by Flex-O-Dyne GmbH -Designed for reproducible, audit-ready, human-centered software. Index: g/certificates.org =================================================================== --- org/certificates.org (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,210 +1,0 @@ -* Flex-O Cross-Platform Signing System - -This document defines how Flex-O performs cryptographic signing across Linux, Windows, and macOS using a single interoperable workflow. - -** Requirements - -- PKCS#7 / CMS signatures -- RSA-4096 -- DER output format (Windows/macOS native) -- No Python crypto libraries -- OS-native signing tooling - -* 1. Key and Certificate Creation (Universal) - -** 1.1 Generate a private key (recommended: RSA-4096) - -#+begin_src bash -openssl genrsa -out mykey.pem 4096 -#+end_src - -** 1.2 Generate an X.509 certificate - -#+begin_src bash -openssl req -new -x509 -key mykey.pem -out mycert.pem -days 3650 -sha256 -#+end_src - -Result: -- mykey.pem → private key -- mycert.pem → certificate (public key) - -* 2. Import Certificates into Each Platform - -** 2.1 Linux - -No import required. Use PEM files directly. - -** 2.2 Windows (PKCS#12 required) - -Create PKCS#12 bundle: - -#+begin_src bash -openssl pkcs12 -export -out flexo.pfx -inkey mykey.pem -in mycert.pem -#+end_src - -Import into user certificate store: - -#+begin_src bash -certutil -user -p PASSWORD -importpfx flexo.pfx -#+end_src - -** 2.3 macOS (PKCS#12) - -#+begin_src bash -openssl pkcs12 -export -out flexo.p12 -inkey mykey.pem -in mycert.pem -#+end_src - -Import: - -#+begin_src bash -security import flexo.p12 -k ~/Library/Keychains/login.keychain-db -#+end_src - - -* 3. Cross-Platform Signing Commands - -** 3.1 Linux (OpenSSL CMS / DER) - -#+begin_src bash -openssl cms -sign \ - -binary \ - -in data.txt \ - -signer mycert.pem \ - -inkey mykey.pem \ - -outform DER \ - -out signature.p7s -#+end_src - -** 3.2 Windows (certutil) - -#+begin_src bash -certutil -sign data.txt signature.p7s -#+end_src - -** 3.3 macOS (`security cms`) - -#+begin_src bash -security cms -S \ - -N "Common Name of Cert" \ - -i data.txt \ - -o signature.p7s -#+end_src - -All three platforms produce binary DER PKCS#7 signatures. - -* 4. Cross-Platform Verification - -** 4.1 Linux (OpenSSL) - -#+begin_src bash -openssl cms -verify \ - -in signature.p7s \ - -inform DER \ - -content data.txt \ - -CAfile mycert.pem \ - -purpose any \ - -out /dev/null -#+end_src - -** 4.2 Windows - -#+begin_src bash -certutil -verify signature.p7s data.txt -#+end_src - -** 4.3 macOS - -#+begin_src bash -security cms -D -i signature.p7s > verified.txt -#+end_src - -* 5. Flex-O Signing Specification - -** 5.1 Key Requirements -- RSA-4096 -- X.509 certificate -- Valid for ≥ 10 years - -** 5.2 Signature Format Requirements -- PKCS#7/CMS -- Binary DER form -- Signing certificate must be embedded - -** 5.3 Verification Requirements -- Must work with OpenSSL CMS -- No dependency on OS certificate stores -- Must accept DER PKCS#7 signatures - -** 5.4 Flex-O Signature Entity Schema -Required fields: -- signed_entity: FlexOID -- signer_id: UUID -- signature_data: base64(PKCS7 DER blob) -- signature_type: "PKCS7-DER" -- certificate_thumbprint: SHA-1 thumbprint -- comment: optional - -** 5.5 Security Assumptions -- Flex-O never stores private keys -- OS handles private key protection -- Only public certificates embedded in signatures - -* 6. Refer to certificates - -** Linux - -#+BEGIN_SRC python -cert_ref = CertificateReference( - platform="LINUX", - identifier="/etc/flexo/certs/mycert.pem", - private_key_path="$HOME/.flexo/mykey.pem", - public_cert_path="/etc/flexo/certs/mycert.pem", -) -backend = create_backend(cert_ref) -signature_bytes = backend.sign(b"hello world") -#+END_SRC - - -** Windows - -#+BEGIN_SRC python -cert_ref = CertificateReference( - platform="WINDOWS", - identifier="E1A2B3C4D5F6A7B8C9D0E1F2A3B4C5D6E7F8A9B", # Thumbprint -) - -backend = create_backend(cert_ref) -signature_bytes = backend.sign(b"hello world") -#+END_SRC - -** MacOS - -#+BEGIN_SRC python -cert_ref = CertificateReference( - platform="MACOS", - identifier="FlexOSigner", # Common Name (CN) - public_cert_path="/Users/enno/certs/FlexOSigner.pem" -) - -backend = create_backend(cert_ref) -signature_bytes = backend.sign(b"hello world") -#+END_SRC - -* 8. Flex-O Signature Entity - -#+begin_src python -@dataclass -class Signature(FlexoEntity): - ENTITY_TYPE = EntityType.OUTPUT - - signed_entity: Optional[FlexOID] = None - signer_id: Optional[UUID] = None - signature_data: str = "" # Base64 of PKCS#7 DER - signature_type: str = "PKCS7-DER" - certificate_thumbprint: str = "" - comment: Optional[str] = None - - @property - def text_seed(self) -> str: - return f"{self.signed_entity}:{self.signer_id}:{self.certificate_thumbprint}" -#+end_src Index: g/transitions.org =================================================================== --- org/transitions.org (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,201 +1,0 @@ -FLEXO ID Decision Flowchart - -We assume: - -flexo_base_id = logical identity - -flexo_version = content evolution - -lifecycle suffix = process state - -fingerprint = canonical JSON hash - -STEP 0 — What is happening? - -You must classify the operation into one of three categories: - -State transition - -Content modification - -Logical duplication / fork - -Everything falls into one of these. - -FLOWCHART -START - | - |-- Is this operation creating a new independent lineage? - | (copy into new catalog? fork? split ownership?) - | - |-- YES --> Generate NEW flexo_base_id - | flexo_version = 1 - | lifecycle_state = Draft - | origin_flexo_id = previous full ID - | END - | - |-- NO - | - |-- Is canonical content different from last saved version? - | (compare normalized JSON, excluding lifecycle metadata) - | - |-- YES --> flexo_base_id stays the same - | flexo_version = previous_version + 1 - | lifecycle_state = Draft - | END - | - |-- NO - | - |-- Is this a lifecycle transition? - | - |-- YES --> flexo_base_id stays the same - | flexo_version stays the same - | lifecycle_state changes - | END - | - |-- NO - | - |-- No ID change required - (metadata only, DB-only operation) - END -Now Let’s Make It Concrete -1️⃣ Logical Duplication (NEW BASE ID) - -Trigger conditions: - -Copy question into new catalog (independent evolution) - -Split catalog into two - -Fork domain for separate governance - -“Save as new” intentionally - -Effect: - -new_base_id = generate() -version = 1 -state = Draft -origin_flexo_id = previous full ID - -Strict rule: - -If future changes must not affect original lineage → new base ID. - -2️⃣ Content Modification (VERSION BUMP) - -Trigger: - -Canonical JSON changed (after normalization). - -Examples: - -Question text edited - -Options reordered (if order matters) - -Points changed - -Domain description changed - -Exam layout changed - -Embedded question snapshot changed - -Effect: - -base_id unchanged -version += 1 -state = Draft - -Strict rule: - -If fingerprint changes → version bump. - -No exceptions. - -3️⃣ Lifecycle Transition (STATE ONLY) - -Trigger: - -Draft → Approved - -Approved → Signed - -Signed → Published - -Published → Obsolete - -Effect: - -base_id unchanged -version unchanged -state changes - -Strict rule: - -Lifecycle does not alter content version. - -Edge Case Handling -Approved entity edited - -Flow: - -Content change detected - -Version bump - -State automatically reset to Draft - -So: - -BASE@001P → edit → BASE@002D - -Clean and auditable. - -Draft → Approved where draft ID is “temporary” - -Your earlier idea suggested generating new ID at approval. - -Under strict logic: - -That is unnecessary and breaks lineage. - -Instead: - -BASE@001D → BASE@001A - -Same logical identity. - -If you must isolate drafts operationally: -Use different database tiers, not different IDs. - -* Decision Matrix Summary - -| Operation | New Base ID | Version++ | State Change | -|-------------------------+-------------+------------+--------------| -| Copy/fork | | reset to 1 | Draft | -| Content edit | | | Draft | -| State promotion | No | No | Yes | -| Metadata-only DB change | | | | - -One Important Constraint - -Fingerprint must exclude: - -- lifecycle_state -- owner_id -- DB storage metadata - -Otherwise state transitions would falsely trigger version bump. - -Fingerprint must represent semantic content only !!! - -Final Integrity Principle - -There must never exist: - -- two different canonical JSONs with same base ID + version -- two identical canonical JSONs with different base ID (unless forked intentionally) - -If those invariants hold, your identity system is mathematically clean. Index: pyproject.toml =================================================================== --- pyproject.toml (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ pyproject.toml (revision d6e75be17bbd83e4a95dcabd999d3cefa40bc8db) @@ -1,63 +1,19 @@ [build-system] -requires = ["hatchling"] -build-backend = "hatchling.build" +requires = ["poetry-core>=1.0.7"] +build-backend = "poetry.core.masonry.api" +[tool.poetry] +name = "flexo" +version = "0.0.1" +authors = ["Enrico Enno Schwass "] +description = "Core library providing unique lifecycle-aware IDs and entity base classes" +readme = "README.md" +license = "MIT" -[project] -name = "flexoentity" -version = "0.0.1" -description = "Core entity and ID framework for the Flex-O ecosystem." -readme = "README.md" -license = { text = "MIT" } +include = [ + {path = 'tests/*.py'}, +] -authors = [ - { name = "Enrico Enno Schwass", email = "ennoausberlin@mac.com" } -] - -requires-python = ">=3.10.6" - -keywords = [ - "flex-o", - "entity", - "id", - "versioning", - "hashing", - "reproducibility" -] - -classifiers = [ - "Development Status :: 5 - Production/Stable", - "Programming Language :: Python :: 3", - "License :: OSI Approved :: MIT License", - "Operating System :: OS Independent" -] - -# Runtime dependencies -dependencies = [] - - -[project.optional-dependencies] -dev = [ - "pytest>=7.1.3" -] - - -[project.urls] -Homepage = "https://gitlab.kokyou.de/enno/flexoentity" -Repository = "https://gitlab.kokyou.de/enno/flexoentity.git" -Documentation = "https://gitlab.kokyou.de/enno/flexoentity/docs" - - -# ───────────── Hatchling configuration ───────────── - -[tool.hatch.build.targets.wheel] -packages = ["flexoentity"] - - -[tool.hatch.build.targets.sdist] -include = [ - "flexoentity/**", - "tests/**", - "README.md", - "pyproject.toml" -] +[tool.poetry.dependencies] +python = "^3.10.6" +pytest = "^7.1.3" Index: sts/conftest.py =================================================================== --- tests/conftest.py (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,135 +1,0 @@ -import pytest -import platform -from pathlib import Path -from datetime import datetime -from flexoentity import Domain, FlexoSignature, DomainManager, EntityRegistry, CompositeBackend, InMemoryBackend -from flexoentity import get_signing_backend, CertificateReference - - -@pytest.fixture -def fixed_datetime(monkeypatch): - class FixedDate(datetime): - @classmethod - def now(cls, tz=None): - return datetime(2025, 11, 1, tzinfo=tz) - monkeypatch.setattr("flexoentity.id_factory.datetime", FixedDate) - return FixedDate - - -@pytest.fixture -def sample_domain(): - domain_id = "PY_ARITHM" - return Domain.with_domain_id(domain_id=domain_id, - fullname="PYTHON_ARITHMETIC", - description="ALL ABOUT ARITHMETIC IN PYTHON") - - -SYSTEM = platform.system() - -@pytest.fixture -def local_backend(): - return InMemoryBackend(Domain) - -@pytest.fixture -def staging_backend(): - return InMemoryBackend(Domain) - -@pytest.fixture -def permanent_backend(): - return InMemoryBackend(Domain) - -@pytest.fixture -def sample_domain_manager(local_backend, staging_backend, permanent_backend): - return DomainManager(local_backend=local_backend, - staging_backend=staging_backend, - permanent_backend=CompositeBackend(authoritative_backend=permanent_backend, - sync_backends=None), - registry=EntityRegistry()) - -# ───────────────────────────────────────────────────────────── -# Basic test data directory + PEM test files -# ───────────────────────────────────────────────────────────── - - -@pytest.fixture(scope="session") -def test_data_dir(): - return Path(__file__).parent / "data" - - -@pytest.fixture(scope="session") -def test_cert(test_data_dir): - return test_data_dir / "testcert.pem" - - -@pytest.fixture(scope="session") -def test_key(test_data_dir): - return test_data_dir / "testkey.pem" - - -# ───────────────────────────────────────────────────────────── -# CertificateReference fixtures for each platform -# ───────────────────────────────────────────────────────────── - -@pytest.fixture(scope="session") -def cert_ref_linux(test_cert, test_key): - """Linux: Uses OpenSSL CMS with PEM cert + PEM private key.""" - return CertificateReference( - platform="LINUX", - identifier=str(test_cert), - private_key_path=str(test_key), - public_cert_path=str(test_cert), - ) - - -@pytest.fixture(scope="session") -def cert_ref_macos(test_cert): - """ - macOS: Uses Keychain identity with Common Name (CN). - The test cert must be imported into the login keychain with CN=FlexOSignerTest. - """ - return CertificateReference( - platform="MACOS", - identifier="FlexOSignerTest", - public_cert_path=str(test_cert), - ) - - -@pytest.fixture(scope="session") -def signing_backend(test_cert, test_key): - """Return the correct backend for the current platform.""" - - if SYSTEM == "Linux": - cert_ref = CertificateReference( - platform="LINUX", - identifier=str(test_cert), - private_key_path=str(test_key), - public_cert_path=str(test_cert), - ) - - elif SYSTEM == "Darwin": - cert_ref = CertificateReference( - platform="MACOS", - identifier="FlexOSignerTest", - public_cert_path=str(test_cert), - ) - - elif SYSTEM == "Windows": - pytest.skip("Windows signing tests not implemented yet") - - else: - pytest.skip(f"Unsupported platform: {SYSTEM}") - - try: - backend = get_signing_backend(cert_ref) - # sanity check: ensures cert exists and command is available - _ = backend.certificate_thumbprint - return backend - except Exception as e: - pytest.skip(f"Backend unavailable or misconfigured: {e}") - - -@pytest.fixture -def sample_signature(sample_domain, cert_ref_linux): - return FlexoSignature.with_domain_id(domain_id="SIG", signed_entity=sample_domain, - certificate_reference=cert_ref_linux, - comment="This is a mock signature") Index: sts/data/testcert.pem =================================================================== --- tests/data/testcert.pem (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,31 +1,0 @@ ------BEGIN CERTIFICATE----- -MIIFazCCA1OgAwIBAgIUChNYApno8zSGa2I1CAigI6Wm5JIwDQYJKoZIhvcNAQEL -BQAwRTEYMBYGA1UEAwwPRmxleE9TaWduZXJUZXN0MQ4wDAYDVQQKDAVGbGV4TzEM -MAoGA1UECwwDRGV2MQswCQYDVQQGEwJERTAeFw0yNTExMjQxMzIwMDhaFw0zNTEx -MjIxMzIwMDhaMEUxGDAWBgNVBAMMD0ZsZXhPU2lnbmVyVGVzdDEOMAwGA1UECgwF -RmxleE8xDDAKBgNVBAsMA0RldjELMAkGA1UEBhMCREUwggIiMA0GCSqGSIb3DQEB -AQUAA4ICDwAwggIKAoICAQDD/z/PnDWjAh4Vpp80MzsxD1K6vgsEV8IEsAa31a5V -4+0BX9RfsmfNkHMR6iwVQStuOJtdBj9VivETx66DTMII6kL0l+HURMu1zv1GNwUz -ZT4eKIbEc+uUI8bAs+Unoi42LWhKcgQV8lsi5DHsFkio7/pRPp06JHCV4fPy4aBY -03/nOUlAC9GGfpai7wtuc8peeVDBIWJWOTrfm8hlekkMBKSAA2uqoA0ONqZzkNXd -ZI5r0Cb0U7yaNISoo0wHG6CsnXO+CXwARwzc0iQdN4CLzmlO3kYTBbiCMnWexNzt -4xmKWDk0JPL2hr4EqqJiAd/Fvu9WQwSjtHQMUlfvafgwK9B8RHBIuXv7W38QVPoR -lk22o9cbfGznmsfGzoHB7wc8PR4msj8USVf4naagG/XyCpnUxXdonatxsbaS0jpi -9ItJ/o9jTxJT5dzVARzKUZnRDR2Mma4JuBf9b+9/VTXwOJAML+4Il43UOC+ZU2Mt -4vlJsnEIHboIgDsYkIb9xh++K62YHxrh/bDOAr91kDovBsMVVCFDOrHh2m1jkPeE -54jBousXe7aKW/sgjwO6g2nDA+hiIE6TffmM5m7b93s7YwuoiGI+S6J3XluEHNRv -rzZ3GRG3YW2wdgfJezm7Et6FL9jS6AmvowGFgTXcJTh1jnxkuBt/U/0cgQtFQR7F -PwIDAQABo1MwUTAdBgNVHQ4EFgQUwQFLZ303o+o6mA/zzCKDtne/jSEwHwYDVR0j -BBgwFoAUwQFLZ303o+o6mA/zzCKDtne/jSEwDwYDVR0TAQH/BAUwAwEB/zANBgkq -hkiG9w0BAQsFAAOCAgEAQyzDHm8iXLCWIimclnG3vhweWf7pNiG9FkNaJo83lfIs -skRB7tfvC1P1KNBSpKxiogPc9iIWMj48oRcRYPhrCLrPApZtViH32MmcFFizECxH -sHjLhy2Xr4HpqW1nF3pGC3BSt0uphiIjjAxFF58Ah/LkUjftMyao6PQX/OZOqzEJ -uIUCmsgZJuGQ0EwwAkua8eP6DyTfemg3YgqgB1BlZET4fQ7mfQpPv6hZgrGWYB5o -FGUPRWhpE6jo18s1zUchFepW3AfYqi2Bt52vhJKu8MLbCCwM+tyjT2jcrXRbP9Gw -dpsjrLkCY2XDBjgRJFlprFUhfPMlMH/XidVke7bBA01jnxxbsOL/FxMLFGyNW1WR -DNbIVgTI9lTwrg88M6erJVPEfjBkZuAAigklMmOCjVAXDORdjvjhzu2HypvLXA/1 -ffX3CS8XDeLDu8i0UpxOVlPr5ax9pNjQ6LtznBLThIQ/3N3NkURLgtbs+gHMG5lY -wNwAbMmCp6wi6QIJEOjKt5j/QEZFklKao1h8BjRvTK0vCrJvbqBRUi9bEbYzP+Z1 -6OQDKsglPjxZE5OgSbgStgYgmiZhjon45BDmn2TejYVcxdrVPIeHBc1E55Q/vrxm -v4BVI1OJgW4Qj8SQTGEwJLWhy2FQvwaOqkjGBtItfC+qtkX2yRupOhtUI3v7f8U= ------END CERTIFICATE----- Index: sts/data/testkey.pem =================================================================== --- tests/data/testkey.pem (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,52 +1,0 @@ ------BEGIN PRIVATE KEY----- -MIIJQwIBADANBgkqhkiG9w0BAQEFAASCCS0wggkpAgEAAoICAQDD/z/PnDWjAh4V -pp80MzsxD1K6vgsEV8IEsAa31a5V4+0BX9RfsmfNkHMR6iwVQStuOJtdBj9VivET -x66DTMII6kL0l+HURMu1zv1GNwUzZT4eKIbEc+uUI8bAs+Unoi42LWhKcgQV8lsi -5DHsFkio7/pRPp06JHCV4fPy4aBY03/nOUlAC9GGfpai7wtuc8peeVDBIWJWOTrf -m8hlekkMBKSAA2uqoA0ONqZzkNXdZI5r0Cb0U7yaNISoo0wHG6CsnXO+CXwARwzc -0iQdN4CLzmlO3kYTBbiCMnWexNzt4xmKWDk0JPL2hr4EqqJiAd/Fvu9WQwSjtHQM -UlfvafgwK9B8RHBIuXv7W38QVPoRlk22o9cbfGznmsfGzoHB7wc8PR4msj8USVf4 -naagG/XyCpnUxXdonatxsbaS0jpi9ItJ/o9jTxJT5dzVARzKUZnRDR2Mma4JuBf9 -b+9/VTXwOJAML+4Il43UOC+ZU2Mt4vlJsnEIHboIgDsYkIb9xh++K62YHxrh/bDO -Ar91kDovBsMVVCFDOrHh2m1jkPeE54jBousXe7aKW/sgjwO6g2nDA+hiIE6TffmM -5m7b93s7YwuoiGI+S6J3XluEHNRvrzZ3GRG3YW2wdgfJezm7Et6FL9jS6AmvowGF -gTXcJTh1jnxkuBt/U/0cgQtFQR7FPwIDAQABAoICAADuGihrDloarXfe1YyS3aoK -75KRSk4X+IS7LRz8N5qSvVIvWTunBhUUpWclXFYxe/pG7H3RhMVsJl64qNxYpecS -7YRpoBm4xdq0A8GsiyrGRTgxawpNnoWSceQCNoksnNmG96K6zcgo4UPWH2KGbIvY -r7BpAqf++kXLz7OMXI1vW4EGZZGXPMRNn3tgdQZuKDywiLCR7vyHwv7Cp33LycLz -1rAA1Cb4IXe6zKlpu4oUaxSZ5UVtjEyTCTofpEG4YNfQ23bAZsU9kRMPMDqvQ73m -GdvU9DIs6cY8ZRB+0KmUVU+xwXu83SZMV2SARsr+RUiT8mDPaps4RQC191Lcht/5 -g3An9VXhk7sRJcKwvAe/nZgY5b9GP7LKcSQtul1L2bp1r3rEIRtKJBoUAxuORbCO -vg6nsu4G2FFbbk9RjxqBt4adP7KxjxBFC0ts7I7NvtxGB11Y7Vkr/za/ifksEyn0 -4somLACVeaU8KgZLuRMvQK/yt2PkQ+sxpup/UOTyoQus2jocw/h/Bk+dJjB8p6lx -Cs0EIynj68Ffd69kq5YKlK0h78pDRe/Bg+VvMPGls0DtuRodQHuggDgHRIbyJasf -yGuWrwluWUDNh1QSkQXX675zvaMPA60WkpHB8fh7KjKPxboeSrFAklSVSeQ3OHDy -kOOSTeZEU+9DU6AZTcdZAoIBAQDxqjGmVge8W8mKpSrRERt3J4RXc/VZlpJDuViD -QC/rTWRnRtURxk4u6SV3S2w6ZyKW/TneYB40AhatfBglZfs7sj14S7qzVCCxk/rD -HU1ybXoINgcSFx+BnlLl3Yob/xWAq90mEK/ded5avH47kq1/u7J8NPu/GavZGepi -kPuRHC1I1yXHfEcwYnmVXYjSy7qj/iylg7pxUvw5bAI9fZsX9Bhbi2Kop+ks7WZx -ArvhCWhuXL/hlJkCRHrdCS5tVXSLjB34u//mRbHXEVRIvO64mJG+Ar/CbsZCBeQ2 -6csf/PmKbJpvVC4rPGHsvOFbHK9NCfZpm6j4gteq968/h1DjAoIBAQDPn5EVDgno -3c/2CPMZ9BaEHWDQODQiCVYMNJusw5u3JxEdgVbjQHRoT11YamAEwFbqdmm23eVZ -vPHcKwuoXdE7s7wKiNnJN1goS8Iw12CiiVccQBy/0HQ3zdZerB/82P0/65AzuIF9 -zJW2jD6yQv7iHH/U0wHzxdlNUMzbUs69qZeOIjScsiUYXxcio0DUW3FWOOzZRW8z -BpP0GIsDpccZ5ZDnIPuQN8NiYBVqPt0tkgxnbHzpc0mE443umVxQ2C1dOUSB9tFs -NSK5ZNshY7gpKyu/aNsZClrFKfVOae1eJ70PXyPglPR94CzyvohHgoewtP4y8pu5 -+V72RXKrQvT1AoIBAQC9IiEneC8nuIJccvW2l/fx4kiOFR/RFKm2PSnL1wFp27EX -rvT0q0nnJur+mVXDw5Hrr5WJt5oLxBKxhexz4IOmbdH0AvjEfrPzpsfbymQhiRK3 -uGkCPHyY+isQ4bjEPng6sscqkmbBJC7kvp7gyuPkkyaWVPUt086N491vdN6d6/p5 -BoD3xgkFEhzrzD9YEsAotrWEeDsDlBn4atK6A0LNlWk2xDjbnEY+tjG1gpk9/xnv -PcGir6SshKHPue7O/NFwoaGXWLNnKeaMVnCrWMpQSH/PwOEJL58Ubv30CDfD+j3m -myamuLBFgEV3fT+2ChYcag6jCoNMs0JM+Pn781BXAoIBAQCY9Kv0f/Aq/uZLElLO -BvVjhSUioJU/kfMzcLpTL5QS/RPt0bBKINzhT76r+UodlfkyJ+Q/lP92+eyQT6H5 -+ou/WO0qMaGITF6E6TL5umH1vApRxKWpZg+IYPsRqeqy4sTHh2onwECdhc+xuWYi -+5o8x9Qg3QauKU6qV+FjnoCyVzNQwcBSxCbx6nPnz69eArPfWnaapj2CsNDk3gPj -vwiL+oSi0biiiYJGghSkvgVBojvDIbhwX7+EyANzCMZqxKe+6waeXE/yUL7xJ+0L -jyRIKAbFr6DhtLAkWoHer8jwOYRdw7BxrirkZPeYKWfKjs4aD2zJ6dNx0dX0xjrx -EpXVAoIBAC8Js0CkPgH82bp7tfekvRLfYUJPCvf8mL2F0gJrTDj2H9D7nrNiOMw6 -y2ekY6t9EE2ivc1KRQsscUv45ZpZDqoxaExdCYtkt4kyYripIFPKrFklFwoe3hCZ -6NJwL9KHjCZ0SRDvImnMvOzkPSt2Ki7EfezptTj4LQdZXOfSY8TR1P07lrU/I+Z6 -eD1AkqJ1xogu//Te7k7et/z3nv89Wo5+9mngW4Wt9+Z18y4LomOyoMfpUFR+3ORN -SLZkMzYOVEXiuTI7vt1XcKbQik9L5DHSQUzbYoYJXBV98S3EYEODZuMZbeohk6lH -COTzylGJM5V2427v9SqnT+ic1kCvejw= ------END PRIVATE KEY----- Index: sts/test_collection.py =================================================================== --- tests/test_collection.py (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,19 +1,0 @@ -from flexoentity import FlexoCollection, Domain - - -def test_collection_basic(): - c = FlexoCollection() - assert len(c) == 0 - - e1 = Domain.with_domain_id("WIP_TEST") - e2 = Domain.with_domain_id("TEST_WIP") - - c.add(e1) - c.add(e2) - - assert len(c) == 2 - assert e1.flexo_id in c - assert c.get(e1.flexo_id) is e1 - - c.remove(e1.flexo_id) - assert len(c) == 1 Index: sts/test_composite_backend.py =================================================================== --- tests/test_composite_backend.py (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,46 +1,0 @@ -from flexoentity.in_memory_backend import InMemoryBackend -from flexoentity.composite_backend import CompositeBackend -from flexoentity.domain import Domain - - -def test_composite_writes_to_all_backends(sample_domain): - primary = InMemoryBackend(Domain) - secondary = InMemoryBackend(Domain) - - backend = CompositeBackend(authoritative_backend=primary, sync_backends=[secondary]) - - backend.save(sample_domain.to_dict()) - - fid = sample_domain.flexo_id - - # both must see the entity - assert primary.load(fid) is not None - assert secondary.load(fid) is not None - - -def test_composite_reads_from_primary_only(sample_domain): - primary = InMemoryBackend(Domain) - secondary = InMemoryBackend(Domain) - - backend = CompositeBackend(authoritative_backend=primary, sync_backends=[secondary]) - - primary.save(sample_domain.to_dict()) - - # secondary has nothing, but composite should still load from primary - fid = sample_domain.flexo_id - loaded = Domain.from_dict(backend.load(fid)) - assert isinstance(loaded, Domain) - - -def test_composite_clear_propagates(sample_domain): - primary = InMemoryBackend(Domain) - secondary = InMemoryBackend(Domain) - - backend = CompositeBackend(authoritative_backend=primary, sync_backends=[secondary]) - - backend.save(sample_domain.to_dict()) - - backend.clear() - - assert primary.load_all() == [] - assert secondary.load_all() == [] Index: sts/test_domain.py =================================================================== --- tests/test_domain.py (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,55 +1,0 @@ -import pytest -from flexoentity import Domain, DomainManager, DuplicateDomainError - - -# --------------------------------------------------------------- -# Setup/Teardown (clear DomainManager between tests) -# --------------------------------------------------------------- - - -@pytest.fixture(autouse=True) -def clear_domain_manager(sample_domain_manager): - sample_domain_manager.clear() - - -# --------------------------------------------------------------- -# Domain creation + registration -# --------------------------------------------------------------- -def test_domain_registration(sample_domain_manager, sample_domain): - - sample_domain_manager.add(sample_domain) - # Manager must know it now - assert sample_domain_manager.get_by_domain_id("PY_ARITHM") is sample_domain - - -# --------------------------------------------------------------- -# Uniqueness: registering the same code twice must fail -# --------------------------------------------------------------- -def test_domain_uniqueness_enforced(sample_domain, sample_domain_manager): - - sample_domain_manager.add(sample_domain) - with pytest.raises(DuplicateDomainError): - sample_domain_manager.add(sample_domain) - - -# --------------------------------------------------------------- -# Lookup by FlexOID -# --------------------------------------------------------------- -def test_lookup_by_oid(sample_domain, sample_domain_manager): - sample_domain_manager.add(sample_domain) - found = sample_domain_manager.get_by_domain_id(sample_domain.domain_id) - assert found is sample_domain - -# --------------------------------------------------------------- -# JSON roundtrip should preserve identity and not regenerate FlexOID -# --------------------------------------------------------------- -def test_domain_json_roundtrip(sample_domain): - sample_data = sample_domain.to_dict() - loaded = Domain.from_dict(sample_data) - - # Same exact instance - assert loaded == sample_domain - - # Same FlexOID - assert str(loaded.flexo_id) == str(sample_domain.flexo_id) - Index: sts/test_flexoid.py =================================================================== --- tests/test_flexoid.py (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,147 +1,0 @@ -#!/usr/bin/env python3 - -# tests/test_flexoid.py -""" -Test suite for id_factory.FlexOID. - -I verify that FlexOID behaves deterministically, validates itself strictly, -and evolves correctly through version and state transitions. -""" - -import re -from datetime import datetime, date -import pytest -from logging import Logger -from flexoentity import FlexOID, canonical_seed, Domain - - -logger = Logger(__file__) - -# ────────────────────────────────────────────── -# Basic construction and validation -# ────────────────────────────────────────────── - -def test_valid_flexoid_parsing(): - fid = FlexOID("GEN-I251101-ABCDEF123456@001D") - assert isinstance(fid, str) - assert fid.domain_id == "GEN" - assert fid.entity_type == "I" - assert fid.date_str == "251101" - assert fid.version == 1 - assert fid.state_code == "D" - assert fid.prefix.endswith("ABCDEF123456") - assert str(fid).endswith("@001D") - - -def test_invalid_flexoid_raises(): - with pytest.raises(ValueError): - FlexOID("INVALIDFORMAT") - with pytest.raises(ValueError): - FlexOID("GEN-I251101-ABCDEF@1D") # bad version width - - -def test_regex_is_strict(): - pat = FlexOID.OID_PATTERN - assert re.match(pat, "GEN-I251101-123456ABCDEF@001A") - assert not re.match(pat, "gen-item251101-123456@001A") # lowercase not allowed - assert not re.match(pat, "GEN-I251101-123456@01A") # wrong digit count - - -# ────────────────────────────────────────────── -# Generation and deterministic hashing -# ────────────────────────────────────────────── -def test_generate_is_not_deterministic(fixed_datetime): - fid1 = FlexOID.generate("GEN", "I", "D", "test content") - fid2 = FlexOID.generate("GEN", "I", "D", "test content") - - assert fid1 != fid2 - assert fid1.domain_id == fid2.domain_id == "GEN" - assert fid1.entity_type == fid2.entity_type == "I" - assert fid1.version == fid2.version == 1 - assert fid1.state_code == fid2.state_code == "D" - -def test_fingerprint_is_stable(): - d1 = Domain.with_domain_id("GEN", fullname="A", description="B") - d2 = Domain.from_dict(d1.to_dict()) - - assert d1.fingerprint == d2.fingerprint - -def test_safe_generate_collision(monkeypatch): - first = FlexOID.generate("GEN", "I", "D", "abc") - - class DummyRepo: - def get(self, key): - if key == str(first): - return True - return None - - repo = DummyRepo() - - fid = FlexOID.safe_generate("GEN", "I", "D", "abc", repo=repo) - - assert isinstance(fid, FlexOID) - assert fid != first - assert fid.state_code == "D" - -# ────────────────────────────────────────────── -# State and version transitions -# ────────────────────────────────────────────── - -def test_with_state_creates_new_instance(): - fid = FlexOID("GEN-I251101-ABCDEF123456@001D") - fid2 = fid.with_state("A") - assert fid != fid2 - assert fid.version == fid2.version - assert fid2.state_code == "A" - assert str(fid2).endswith("@001A") - - -def test_next_version_increments_properly(): - fid = FlexOID("GEN-I251101-ABCDEF123456@001A") - fid2 = FlexOID.next_version(fid) - assert fid2.version == 2 - assert fid2.state_code == fid.state_code - assert fid.prefix == fid2.prefix - - -def test_from_oid_and_version_sets_exact_version(): - fid = FlexOID("GEN-I251101-ABCDEF123456@001A") - fid2 = FlexOID.from_oid_and_version(fid, 10) - assert fid2.version == 10 - assert fid2.state_code == "A" - - -def test_clone_new_base_starts_at_one(fixed_datetime): - fid = FlexOID.clone_new_base("GEN", "I", "D", "clone text") - assert fid.version == 1 - assert fid.state_code == "D" - - -# ────────────────────────────────────────────── -# Canonical seed behavior -# ────────────────────────────────────────────── - -def test_canonical_seed_for_strings(): - s1 = " Hello world " - s2 = "Hello world" - assert canonical_seed(s1) == canonical_seed(s2) - -# ────────────────────────────────────────────── -# Parsed structure and representations -# ────────────────────────────────────────────── - -def test_parsed_returns_expected_dict(): - fid = FlexOID("GEN-I251101-ABCDEF123456@007S") - data = fid.to_dict() - assert data["domain_id"] == "GEN" - assert data["entity_type"] == "I" - assert data["version"] == 7 - assert data["state"] == "S" - assert isinstance(data["date"], date) - - -def test_repr_is_readable(): - fid = FlexOID("GEN-I251101-ABCDEF123456@001D") - s = repr(fid) - assert "FlexOID" in s - assert "GEN-I" in s Index: sts/test_from_strings.py =================================================================== --- tests/test_from_strings.py (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,140 +1,0 @@ -import pytest -from flexoentity.id_factory import FlexOID - - -# ───────────────────────────────────────────────────────────────────────────── -# Valid construction via from_strings -# ───────────────────────────────────────────────────────────────────────────── -def test_from_strings_valid(): - oid = FlexOID.from_strings( - domain_id="GEN_GENERIC", - entity_type="I", - date="251101", - hash_part="A1B2C3D4E5F6", - version="001", - state="D", - ) - - assert isinstance(oid, FlexOID) - assert str(oid) == "GEN_GENERIC-I251101-A1B2C3D4E5F6@001D" - - assert oid.domain_id == "GEN_GENERIC" - assert oid.entity_type == "I" - assert oid.date_str == "251101" - assert oid.hash_part == "A1B2C3D4E5F6" - assert oid.version == 1 - assert oid.state_code == "D" - - -# ───────────────────────────────────────────────────────────────────────────── -# from_dict should delegate to from_strings -# ───────────────────────────────────────────────────────────────────────────── -def test_from_dict_valid(): - data = { - "domain_id": "AF", - "entity_type": "C", - "date": "250101", - "hash": "ABCDEF123456", - "version": "003", - "state": "A", - } - - oid = FlexOID.from_dict(data) - - assert isinstance(oid, FlexOID) - assert str(oid) == "AF-C250101-ABCDEF123456@003A" - - -# ───────────────────────────────────────────────────────────────────────────── -# Missing dict fields must fail loudly -# ───────────────────────────────────────────────────────────────────────────── -def test_from_dict_missing_fields(): - incomplete = { - "domain": "AF", - "entity_type": "C", - # missing: date, hash, version, state - } - - with pytest.raises(ValueError): - FlexOID.from_dict(incomplete) - - -# ───────────────────────────────────────────────────────────────────────────── -# Invalid strings should be rejected in from_strings -# ───────────────────────────────────────────────────────────────────────────── -@pytest.mark.parametrize("domain", ["geN", "GEN!", "GEN-", "", None]) -def test_from_strings_invalid_domain(domain): - with pytest.raises(ValueError): - FlexOID.from_strings( - domain_id=domain, - entity_type="I", - date="241001", - hash_part="AABBCCDDEEFF", - version="001", - state="D", - ) - - -@pytest.mark.parametrize("etype", ["", "ITEM", "i", "aa"]) -def test_from_strings_invalid_entity_type(etype): - with pytest.raises(ValueError): - FlexOID.from_strings( - domain_id="GEN", - entity_type=etype, - date="241001", - hash_part="AABBCCDDEEFF", - version="001", - state="D", - ) - - -@pytest.mark.parametrize("date", ["20241001", "2410", "ABCDEF", None]) -def test_from_strings_invalid_date(date): - with pytest.raises(ValueError): - FlexOID.from_strings( - domain_id="GEN", - entity_type="I", - date=date, - hash_part="AABBCCDDEEFF", - version="001", - state="D", - ) - - -@pytest.mark.parametrize("hash_part", ["abc123", "ZZZZZZZZ", "GHIJKL", "", None]) -def test_from_strings_invalid_hash(hash_part): - with pytest.raises(ValueError): - FlexOID.from_strings( - domain_id="GEN", - entity_type="I", - date="241001", - hash_part=hash_part, - version="001", - state="D", - ) - - -@pytest.mark.parametrize("version", ["1", "01", "1000", "0A1", "abc", None]) -def test_from_strings_invalid_version(version): - with pytest.raises(ValueError): - FlexOID.from_strings( - domain_id="GEN", - entity_type="I", - date="241001", - hash_part="AABBCCDDEEFF", - version=version, - state="D", - ) - - -@pytest.mark.parametrize("state", ["d", "DD", "1", "", None]) -def test_from_strings_invalid_state(state): - with pytest.raises(ValueError): - FlexOID.from_strings( - domain_id="GEN", - entity_type="I", - date="241001", - hash_part="AABBCCDDEEFF", - version="001", - state=state, - ) Index: sts/test_id_lifecycle.py =================================================================== --- tests/test_id_lifecycle.py (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,161 +1,0 @@ -import pytest -from flexoentity import FlexOID, FlexoEntity, EntityState - - -def test_initial_state(sample_domain): - assert sample_domain.state == EntityState.DRAFT - assert sample_domain.flexo_id.version == 1 - assert FlexoEntity.verify_integrity(sample_domain) - -def test_approval_does_not_bump_version(sample_domain): - q = sample_domain - q.approve() - assert q.state == EntityState.APPROVED - assert q.flexo_id.version == 1 - -def test_signing_does_not_bump_version(sample_domain): - q = sample_domain - q.approve() - before = q.flexo_id - q.sign() - after = q.flexo_id - - # state changed - assert q.state == EntityState.APPROVED_AND_SIGNED - - # version unchanged - assert before.version == after.version - - # only suffix letter differs - assert before.prefix == after.prefix - assert before.state_code == "A" - assert after.state_code == "S" - - -def test_publish_does_not_bump_version(sample_domain): - q = sample_domain - q.approve() - q.sign() - v_before = q.flexo_id.version - q.publish() - assert q.state == EntityState.PUBLISHED - assert q.flexo_id.version == v_before - - -def test_modify_content_changes_fingerprint(sample_signature): - sample_signature.comment += "Rephrased content" # simulate text change - changed = sample_signature._update_fingerprint() - assert changed - - -def test_no_version_bump_on_draft_edits(sample_signature): - sample_signature.comment = "Minor draft edit" - sample_signature._update_fingerprint() - assert sample_signature.flexo_id.version == 1 - - -def test_version_bump_after_edit_and_sign(sample_signature): - sample_signature.approve() - v1 = str(sample_signature.flexo_id) - sample_signature.comment = "Changed comment" - sample_signature.sign() - assert str(sample_signature.flexo_id) != v1 - - -def test_integrity_check_passes_and_fails(sample_signature): - sample_signature.approve() - assert FlexoEntity.verify_integrity(sample_signature) - - # simulate tampering - sample_signature.comment = "Tampered text" - assert not FlexoEntity.verify_integrity(sample_signature) - - -def test_obsolete_state(sample_domain): - q = sample_domain - q.approve() - q.sign() - q.publish() - q.obsolete() - assert q.state == EntityState.OBSOLETE - - -def test_clone_new_base_resets_lineage(sample_domain): - q = sample_domain - q.approve() - q.sign() - q.publish() - q.obsolete() - old_id = str(q.flexo_id) - q.clone_new_base() - assert str(q.flexo_id) != old_id - assert q.state == EntityState.DRAFT - assert q.flexo_id.version == 1 - -def test_clone_new_base_sets_origin(sample_domain): - q = sample_domain - q.approve() - q.sign() - q.publish() - q.obsolete() - old_id = str(q.flexo_id) - q.clone_new_base() - assert q.origin == old_id - assert q.state == EntityState.DRAFT - assert q.flexo_id.version == 1 - assert q.flexo_id != old_id - -def test_mass_version_increments_until_obsolete(sample_domain): - q = sample_domain - q.approve() - for _ in range(FlexOID.MAX_VERSION - 1): - q.bump_version() - - # Next one must raise - with pytest.raises(RuntimeError, match="mark obsolete"): - q.bump_version() - -def test_fork_creates_independent_domain(sample_domain): - original = sample_domain - - forked = original.fork(domain_id="DST") - - # ─── identity ─────────────────────────────────────────────────────── - assert forked is not original - assert isinstance(forked.flexo_id, FlexOID) - assert forked.flexo_id != original.flexo_id - - # ─── lifecycle reset ──────────────────────────────────────────────── - assert forked.state == EntityState.DRAFT - assert forked.version == 1 - - # ─── provenance ───────────────────────────────────────────────────── - assert forked.origin == str(original.flexo_id) - - # ─── domain reassigned ────────────────────────────────────────────── - assert forked.domain_id == "DST" - -def test_fork_does_not_mutate_original(sample_domain): - original = sample_domain - original_id = original.flexo_id - original_fp = original.fingerprint - - _ = original.fork(domain_id="DST") - - assert original.flexo_id == original_id - assert original.fingerprint == original_fp - assert original.origin is None - -def test_fork_does_not_shadow_state(sample_domain): - forked = sample_domain.fork(domain_id="DST") - - # No instance attribute allowed - assert "state" not in forked.__dict__ - - # Property must still work - assert forked.state == EntityState.DRAFT - -def test_fork_without_origin(sample_domain): - forked = sample_domain.fork(domain_id="DST", keep_origin=False) - - assert forked.origin is None Index: sts/test_id_stress.py =================================================================== --- tests/test_id_stress.py (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,100 +1,0 @@ -""" -Stress tests for the Flex-O ID lifecycle. -Focus: collision avoidance, version ceiling, reproducibility. -""" - -import copy -import logging -import random - -import pytest -from uuid import uuid4 -from flexoentity import FlexOID, EntityType, EntityState, FlexoSignature - -logger = logging.getLogger(__name__) - -def test_bulk_generation_uniqueness(sample_domain): - """ - Generate 100,000 IDs and ensure uniqueness using safe_generate(). - If a collision occurs, safe_generate() must resolve it automatically - via salt + date adjustment. - """ - entity_type = EntityType.ITEM - estate = EntityState.DRAFT - seeds = [f"question {i}" for i in range(100000)] - - # Simulate a simple in-memory repository for collision detection - repo = {} - - def repo_get(oid_str): - return repo.get(str(oid_str)) - - # Generate IDs using safe_generate - ids = [] - for seed in seeds: - oid = FlexOID.safe_generate(sample_domain.domain_id, entity_type.value, - estate.value, seed, repo=repo) - assert isinstance(oid, FlexOID) - ids.append(str(oid)) - repo[str(oid)] = oid # register for future collision detection - - unique_count = len(set(ids)) - total_count = len(ids) - collisions = total_count - unique_count - - logger.info(f"Generated {total_count} IDs ({collisions} collisions handled).") - - # Assert that safe_generate avoided duplicates - assert total_count == unique_count, f"Unexpected duplicate IDs ({collisions} found)" - - # Sanity check: IDs should look canonical - # assert all(id_str.startswith("GENERIC") for id_str in ids) - assert all("@" in id_str for id_str in ids) - -# def test_id_generation_is_deterministic(sample_domain): -# """ -# Generating the same entity twice with same inputs yields identical ID. -# (No runtime disambiguation; IDs are deterministic by design.) -# """ -# entity_type = EntityType.ITEM -# estate = EntityState.DRAFT -# text = "identical question text" - -# id1 = FlexOID.generate(sample_domain.domain_id, entity_type.value, estate.value, text) -# id2 = FlexOID.generate(sample_domain.domain_id, entity_type.value, estate.value, text) -# # IDs must be identical because generation is deterministic -# assert id1 == id2 - -def test_massive_lifecycle_simulation(cert_ref_linux, sample_domain): - """ - Generate 100 random FlexoSignatures, mutate content, run through lifecycle, - and ensure all FlexOIDs are unique and valid. - """ - entities = [] - - for i in range(100): - sig = FlexoSignature.with_domain_id( - domain_id="SIGTEST", - signed_entity=sample_domain.flexo_id, - signer_id=uuid4(), - certificate_reference=cert_ref_linux, - comment=f"Initial signature #{i}" - ) - entities.append(sig) - - # Mutate + lifecycle transitions - for i, e in enumerate(entities): - # CONTENT CHANGE → fingerprint changes → hash → FlexOID.prefix changes - e.comment += f" updated-{i}" - e._update_fingerprint() - - # lifecycle transitions - e.approve() - if random.random() > 0.3: - e.sign() - if random.random() > 0.6: - e.publish() - - # Check ID uniqueness - flexoids = [str(e.flexo_id) for e in entities] - assert len(flexoids) == len(set(flexoids)), "Duplicate FlexOIDs detected" Index: sts/test_in_memory_backend.py =================================================================== --- tests/test_in_memory_backend.py (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,41 +1,0 @@ -import pytest -from flexoentity.in_memory_backend import InMemoryBackend -from flexoentity.domain import Domain - - - - -def test_save_and_load_roundtrip(local_backend, sample_domain): - local_backend.save(sample_domain.to_dict()) - - loaded_dict = local_backend.load(sample_domain.flexo_id) - loaded = Domain.from_dict(loaded_dict) - - assert isinstance(loaded, Domain) - assert loaded == sample_domain - -def test_update_overwrites_entity(local_backend, sample_domain): - local_backend.save(sample_domain.to_dict()) - - # change something - sample_domain.description = "UPDATED DESC" - local_backend.update(sample_domain.to_dict()) - - loaded = Domain.from_dict(local_backend.load(sample_domain.flexo_id)) - assert loaded.description == "UPDATED DESC" - - -def test_delete_removes_entity(local_backend, sample_domain): - local_backend.save(sample_domain.to_dict()) - local_backend.delete(sample_domain.flexo_id) - - assert local_backend.load(sample_domain.flexo_id) is None - assert local_backend.load_all() == [] - - -def test_clear_removes_all(local_backend, sample_domain): - local_backend.save(sample_domain.to_dict()) - local_backend.clear() - - assert local_backend.load(sample_domain.flexo_id) is None - assert local_backend.load_all() == [] Index: sts/test_json_file_backend.py =================================================================== --- tests/test_json_file_backend.py (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,34 +1,0 @@ -#!/usr/bin/env python3 - -from flexoentity.domain import Domain -from flexoentity.json_file_backend import JsonFileBackend - - -def make_domain(domain_id="PY_ARITHM"): - return Domain.with_domain_id( - subtype="Domain", - domain_id=domain_id, - fullname="PYTHON_ARITHMETIC", - description="ALL ABOUT ARITHMETIC IN PYTHON", - classification="UNCLASSIFIED", - ) - - -def test_jsonfile_roundtrip(tmp_path): - path = tmp_path / "domains.json" - backend = JsonFileBackend(Domain, path) - - dom1 = make_domain("PY_ARITHM") - dom2 = make_domain("PY_STRINGS") - - backend.save(dom1.to_dict()) - backend.save(dom2.to_dict()) - backend.flush_to_file() - - # new instance simulating a reload - backend2 = JsonFileBackend(Domain, path) - backend2.load_from_file() - loaded_dicts = backend2.load_all() - domains = [Domain.from_dict(d) for d in loaded_dicts] - ids = sorted(d.domain_id for d in domains) - assert ids == ["PY_ARITHM", "PY_STRINGS"] Index: sts/test_persistance_integrity.py =================================================================== --- tests/test_persistance_integrity.py (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,58 +1,0 @@ -""" -Persistence and integrity verification tests for Flex-O entities. -Ensures fingerprints survive JSON export/import and detect tampering. -""" -import json -import pytest - -from flexoentity import Domain - - -@pytest.fixture -def approved_domain(sample_domain): - """Provide a fully approved and published Domain for persistence tests.""" - sample_domain.approve() - sample_domain.sign() - sample_domain.publish() - return sample_domain - - -def test_json_roundtrip_preserves_integrity(approved_domain): - """ - Export to JSON and reload — ensure fingerprints and signatures remain valid. - """ - json_str = approved_domain.to_json() - loaded = Domain.from_json(json_str) - - # Fingerprint and state should match — integrity must pass - assert Domain.verify_integrity(loaded) - - # Metadata should be preserved exactly - assert approved_domain.fingerprint == loaded.fingerprint - assert approved_domain.flexo_id == loaded.flexo_id - assert loaded.state == approved_domain.state - - -def test_json_tampering_detection(approved_domain): - """Tampering with content should invalidate fingerprint verification.""" - json_str = approved_domain.to_json() - tampered = json.loads(json_str) - tampered["content"]["fullname"] = "Tampered content injection" - tampered_json = json.dumps(tampered) - - loaded = Domain.from_json(tampered_json) - assert not Domain.verify_integrity(loaded) - - -def test_json_file_corruption(approved_domain, tmp_path): - """Simulate file corruption — integrity check must fail.""" - file = tmp_path / "question.json" - json_str = approved_domain.to_json() - file.write_text(json_str) - - # Corrupt the file (simulate accidental byte modification) - corrupted = json_str.replace("ARITHMETIC", "ARITHM") - file.write_text(corrupted) - - loaded = Domain.from_json(file.read_text()) - assert not Domain.verify_integrity(loaded) Index: sts/test_signing.py =================================================================== --- tests/test_signing.py (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,63 +1,0 @@ -import platform -import pytest -from uuid import uuid4 -from flexoentity import FlexOID, EntityState, EntityType, FlexoSignature, get_signing_backend - -def test_sign_and_verify_linux(signing_backend): - data = b"Hello Flex-O signing!" - - signature = signing_backend.sign(data) - assert isinstance(signature, bytes) - assert len(signature) > 20 # sanity check - - assert signing_backend.verify(data, signature) is True - -def test_sign_and_verify_macos(signing_backend): - data = b"Hello Flex-O signing!" - - signature = signing_backend.sign(data) - assert isinstance(signature, bytes) - assert len(signature) > 20 # sanity check - - assert signing_backend.verify(data, signature) is True - -def test_verify_fails_with_wrong_data(signing_backend): - data = b"Original Data" - wrong_data = b"Tampered Data" - - signature = signing_backend.sign(data) - - assert signing_backend.verify(data, signature) == True - assert signing_backend.verify(wrong_data, signature) == False - -def test_verify_fails_with_invalid_signature(signing_backend): - data = b"Hello world" - invalid_sig = b"\x00\x01\x02garbagepkcs7data" - - assert signing_backend.verify(data, invalid_sig) is False - -def test_signature_entity_create_and_verify(signing_backend): - entity_id = FlexOID.safe_generate( - domain_id="TEST", - entity_type=EntityType.ATTESTATION.value, - state=EntityState.DRAFT.value, - text="abc", - version=1, - ) - signer = uuid4() - data = b"Hello Entity Signing" - - sig = FlexoSignature.create_signed( - data=data, - entity=entity_id, - signer_id=signer, - backend=signing_backend, - ) - - assert isinstance(sig.signature_data, str) - assert sig.signature_type == "PKCS7-DER" - assert sig.signed_entity == entity_id - assert sig.signer_id == signer - assert sig.certificate_thumbprint != "" - - assert sig.verify(data, signing_backend) is True Index: sts/test_sqlite_backend.py =================================================================== --- tests/test_sqlite_backend.py (revision c1144fd13d1332dc1a70a49f1b21fe91f5c7565f) +++ (revision ) @@ -1,20 +1,0 @@ -import sqlite3 -from flexoentity.domain import Domain -from flexoentity import SQLiteEntityBackend - - -def test_sqlite_roundtrip(tmp_path, sample_domain): - db_path = tmp_path / "domains.db" - conn = sqlite3.connect(db_path) - conn.row_factory = sqlite3.Row - - backend = SQLiteEntityBackend(Domain, conn, table_name="domains") - - backend.save(sample_domain.to_dict()) - - loaded = backend.load(sample_domain.flexo_id) - assert loaded == sample_domain.to_dict() - - all_loaded = backend.load_all() - assert len(all_loaded) == 1 - assert Domain.from_dict(all_loaded[0]).domain_id == "PY_ARITHM"