shopdb-flask/shopdb/api/__init__.py

"""Public API namespace exposed to plugins.

Plugin authors at sister sites import from this module. The contract is
locked in ADR-001 and versioned per ADR-002. Helpers added here become
part of the platform contract; bumps follow ADR-002 rules.

Currently exposed:
- audit_log: record an audit log entry with consistent schema
- resolve_asset_position: compute an asset's resolved map position

Setting helpers are exposed via BasePlugin instance methods
(plugin.get_setting, plugin.set_setting), not from this namespace.
"""

from typing import Any, Dict, Optional

# -- Plugin contract surface (ADR-001, versioned per ADR-002) ----------------
# Everything a plugin is allowed to import from the core lives here. Plugins
# import these from `shopdb.api`, never from internal paths like
# `shopdb.core.models.*` or `shopdb.extensions`. The contract test
# (tests/test_plugin_contract.py) enforces this. Adding a name here is an
# additive (minor) contract change; removing one is breaking (major).

# Infrastructure
from shopdb.extensions import db, cache

# Model base classes for declaring plugin tables
from shopdb.core.models.base import BaseModel, AuditMixin

# Core domain models plugins legitimately reference (the asset contract)
from shopdb.core.models import (
    Asset,
    AssetType,
    AssetStatus,
    Vendor,
    Model,
    Communication,
    CommunicationType,
    Location,
    Setting,
    AuditLog,
    Application,
    AppVersion,
    Script,
    OperatingSystem,
    AssetRelationship,
    RelationshipType,
    User,
    Role,
    SupportTeam,
)
# Display-role mapping (which kiosk shows which surface). On the surface
# because a plugin reporting on displays has no other way to name what a
# display IS - the role lives in core, not in any plugin.
from shopdb.core.models.dashboarddefault import (
    DashboardDefault,
    DISPLAY_ROLES,
    DISPLAY_ROLE_PATHS,
    normalize_display_role,
)

# Response + pagination helpers for plugin API blueprints
from shopdb.utils.responses import (
    success_response,
    error_response,
    paginated_response,
    ErrorCodes,
)
from shopdb.utils.pagination import get_pagination_params, paginate_query

# Authorization decorators for gating plugin write routes
from shopdb.utils.authz import (require_permission, require_any_permission,
                                require_role, caller_holds, caller_is_role)

# Service-token authorization for unattended plugin endpoints (collector,
# GE-Enforce fetch, ...): checks a scoped managed token without exposing token
# internals.
from shopdb.utils.apitoken_auth import (
    service_token_authorized, authorized_service_token,
    mint_service_token, list_service_tokens, revoke_service_tokens,
)

# Collector-token authorization for agent-facing plugin routes that live outside
# the generic collector blueprint (e.g. the endpoints command claim/result
# callbacks). Reuses the same per-plugin collector key + managed-token check the
# ingest route uses, so a plugin never imports core collector internals.
def collector_authorized(pluginname: str):
    """Authorize an agent-facing route on a plugin's collector token.

    Returns None when the caller presented a valid collector credential for
    this plugin (the per-plugin env key or a collector.ingest[.<plugin>]-scoped
    managed token), else the error_response to return. Lets a plugin gate an
    agent callback on the same key and kill switch as its ingest route without
    reaching into core collector internals.

    Imported lazily: shopdb.core.api imports back into shopdb.api (send_upload),
    so a module-level import here would be a circular import at load time.
    """
    from shopdb.core.api.collector import _check_collector_auth, _plugin_api_key
    return _check_collector_auth(_plugin_api_key(pluginname), pluginname)


def application_package_location():
    """Where application installers live, and the URL prefix that serves them.

    Returns (directory, urlprefix). A plugin that registers an installable
    payload (the endpoint agent bundle, for example) must put the file exactly
    where the applications API expects it and build installpath the same way,
    or update_app cannot resolve the download. Reconstructing either by hand
    lets the two drift silently, so both come from one place.

    Imported lazily for the same reason as collector_authorized above:
    shopdb.core.api imports back into shopdb.api, so a module-level import
    would be circular at load time.
    """
    from shopdb.core.api.applications import (
        _apppackage_dir, APP_PACKAGE_URL_PREFIX)
    return _apppackage_dir(), APP_PACKAGE_URL_PREFIX


def script_body_location():
    """Where script bodies live, and the URL prefix that serves them.

    Returns (directory, urlprefix). Same reasoning as
    application_package_location: a plugin that seeds a script into the
    library must write the body where the download route reads it from, and
    reconstructing that path by hand lets the two drift apart silently.

    Imported lazily to avoid the shopdb.core.api -> shopdb.api cycle.
    """
    from shopdb.core.api.scripts import _script_dir, SCRIPT_URL_PREFIX
    return _script_dir(), SCRIPT_URL_PREFIX


# Serving an uploaded file safely. A plugin that serves user-supplied bytes must
# not have to remember the headers that keep an SVG or an HTML file from running
# as script in this origin - see shopdb/utils/uploads.py.
from shopdb.utils.uploads import send_upload, harden_upload_response

# Caller IP for rate limiting and audit. On the surface so a plugin does not
# reimplement it: the obvious hand-rolled version reads X-Forwarded-For, which
# the caller supplies, so a rate limit keyed on it can be rotated away to
# nothing. See shopdb/utils/clientip.py.
from shopdb.utils.clientip import client_ip, strip_port

# Strict date parsing. On the contract surface because every plugin that stores
# a date needs it and the two that rolled their own both swallowed the error.
from shopdb.utils.queryargs import QueryArgError, intarg, intargs

# One badge vocabulary for the site: both kiosks read the same physical badges.
from shopdb.utils.badges import BadgeShapeError, badge_kind, normalise_badge

from shopdb.utils.datevalues import (
    DateValueError,
    parse_date,
    parse_date_fields,
)

# Import-mode helpers: preserve legacy timestamps during a bulk data import
from shopdb.utils.import_mode import (
    apply_import_timestamps,
    import_mode_active,
    parse_import_datetime,
)

# Dualpath single-machine collapse (a dual-bay pair is one physical machine)
from shopdb.core.services.dualpath import (
    resolve_dualpath_pairs,
    dualpath_single_machine_enabled,
)

# Bulk-import extension point: a plugin registers a virtual column on an
# importable CSV table plus a post-write row hook, so a relationship a site
# would otherwise hand-build (a PC's machine number -> a controls link) can be
# written in the row itself. Core never learns what the column means.
from shopdb.core.services.csvimport import (
    register_import_hook,
    RowWarning,
)

# Legacy employee directory lookup (read-only) used by notifications
from shopdb.utils.employee_db import employee_connection
from shopdb.utils.mailer import send_email, send_alert, send_webhook

# CMMC USB check-in/out database (read-write) used by the usb plugin
from shopdb.utils.cmmc_usb_db import cmmc_usb_connection


# Reversible bulk-import journal. A plugin that drives a bulk import (a site
# asset adoption, an integration sync) records the primitive changes it makes so
# the whole batch can be undone. ImportBatch is the identity approval, write, and
# undo all reference; ImportJournalEntry is one reversible mutation. Exposed so a
# plugin's import UI can list applied batches and reverse one without reaching
# into core internals. See the PLUGIN-HOOKS wiki page.
from shopdb.core.models import (
    ImportBatch,
    ImportJournalEntry,
    BATCH_APPLIED,
    BATCH_PARTIAL,
    BATCH_UNDONE,
    OP_UPDATE,
    OP_INSERT,
)


def audit_log(
    action: str,
    entitytype: str,
    entityid: int = None,
    entityname: str = None,
    changes: Dict = None,
    details: Dict = None,
) -> AuditLog:
    """Record an audit log entry with the framework's standard schema.

    Plugins call this to record state changes on their own assets in a
    way that is consistent with core auditing. The function delegates to
    AuditLog.log() which already captures the current user, IP address,
    and user agent from the Flask request context.

    Args:
        action: Action verb in past tense ('created', 'updated', 'deleted')
        entitytype: Class name of the entity affected ('Computer', 'Printer')
        entityid: Primary key of the entity
        entityname: Human-readable identifier (hostname, asset number)
        changes: Dict with 'before' and 'after' snapshots for updates
        details: Arbitrary additional context

    Returns:
        The created AuditLog instance, already committed to the DB.
    """
    return AuditLog.log(
        action=action,
        entitytype=entitytype,
        entityid=entityid,
        entityname=entityname,
        changes=changes,
        details=details,
    )


# Cap how deep the relationship-walk traverses before giving up. ADR-001
# specifies a max walk depth of 3 to bound the work per request, with a
# visited-set guarding against cycles. Past this depth the walk treats the
# next hop as if it had no position to contribute.
_POSITION_WALK_MAX_DEPTH = 3

# Relationship type names whose edges are eligible for the inheritance walk
# when inheritsposition is true on the edge. Ordered by priority per
# ADR-001 ("partof first, then controls"). Edges of other types are never
# followed even if inheritsposition is true.
_INHERITABLE_TYPES = ('partof', 'controls')


def _walk_related_for_position(asset, visited, depth):
    """Recursive helper for resolve_asset_position relationship walk. Returns
    a (mapx, mapy) tuple from the first related asset whose position
    resolves, or None. Visited tracks assetids already explored to break
    cycles."""
    if depth >= _POSITION_WALK_MAX_DEPTH:
        return None
    aid = getattr(asset, 'assetid', None)
    if aid is None or aid in visited:
        return None
    visited.add(aid)

    edges = []
    for rel in list(getattr(asset, 'outgoing_relationships', []) or []):
        edges.append((rel, getattr(rel, 'targetasset', None)))
    for rel in list(getattr(asset, 'incoming_relationships', []) or []):
        edges.append((rel, getattr(rel, 'sourceasset', None)))

    def _priority(edge):
        rel = edge[0]
        rtype = getattr(rel, 'relationshiptype', None)
        type_name = getattr(rtype, 'relationshiptype', '') if rtype else ''
        try:
            return _INHERITABLE_TYPES.index(type_name)
        except ValueError:
            return len(_INHERITABLE_TYPES)

    edges.sort(key=_priority)

    for rel, neighbor in edges:
        if neighbor is None:
            continue
        if not getattr(rel, 'inheritsposition', False):
            continue
        if not getattr(rel, 'isactive', True):
            continue
        rtype = getattr(rel, 'relationshiptype', None)
        type_name = getattr(rtype, 'relationshiptype', '') if rtype else ''
        if type_name not in _INHERITABLE_TYPES:
            continue

        n_mapx = getattr(neighbor, 'mapx', None)
        n_mapy = getattr(neighbor, 'mapy', None)
        if n_mapx is not None and n_mapy is not None:
            # The neighbour's LEVEL travels with its coordinates. Returning the
            # pair alone would leave the caller drawing a machine's position on
            # whatever level the PC that borrowed it happens to claim.
            return (n_mapx, n_mapy, getattr(neighbor, 'levelid', None))

        recursed = _walk_related_for_position(neighbor, visited, depth + 1)
        if recursed is not None:
            return recursed
    return None


def resolve_asset_position(asset) -> Optional[Dict[str, Any]]:
    """Compute the resolved map position for an asset.

    Per ADR-001, position resolution follows this priority chain:
      1. Asset-specific override (asset.mapx, asset.mapy)
      2. Walk relationships where inheritsposition is true on edges of type
         partof or controls (partof first), depth-limited and cycle-safe
      3. Asset's location coords (asset.location.mapx, .mapy)
      4. None (asset is unplaced, rendered in a tray)

    Returns a dict {'mapx', 'mapy', 'levelid', 'positionsource'} where
    positionsource is one of 'self', 'related', 'location'. Returns None when no
    priority yields coordinates.

    LEVELID COMES FROM WHICHEVER SOURCE SUPPLIED THE COORDINATES, not from the
    asset (ADR-017). A PC with no position of its own that inherits from the
    machine it controls is at the MACHINE's coordinates on the MACHINE's level;
    using the PC's own level - which may be null, or may be a different building
    entirely - would draw those coordinates on the wrong drawing, and the result
    looks entirely reasonable.
    """
    mapx = getattr(asset, 'mapx', None)
    mapy = getattr(asset, 'mapy', None)
    if mapx is not None and mapy is not None:
        return {'mapx': mapx, 'mapy': mapy,
                'levelid': getattr(asset, 'levelid', None),
                'positionsource': 'self'}

    related = _walk_related_for_position(asset, set(), 0)
    if related is not None:
        return {'mapx': related[0], 'mapy': related[1], 'levelid': related[2],
                'positionsource': 'related'}

    location = getattr(asset, 'location', None)
    if location is not None:
        loc_mapx = getattr(location, 'mapx', None)
        loc_mapy = getattr(location, 'mapy', None)
        if loc_mapx is not None and loc_mapy is not None:
            return {
                'mapx': loc_mapx,
                'mapy': loc_mapy,
                'levelid': getattr(location, 'levelid', None),
                'positionsource': 'location',
            }

    return None


__all__ = [
    # Helpers
    'audit_log',
    'resolve_asset_position',
    'resolve_dualpath_pairs',
    'dualpath_single_machine_enabled',
    # Bulk-import extension point
    'register_import_hook',
    'RowWarning',
    # Infrastructure
    'db',
    'cache',
    # Model bases
    'BaseModel',
    'AuditMixin',
    # Core models
    'Asset',
    'AssetType',
    'AssetStatus',
    'Vendor',
    'Model',
    'Communication',
    'CommunicationType',
    'Location',
    'Setting',
    'AuditLog',
    'Application',
    'AppVersion',
    'Script',
    'OperatingSystem',
    'AssetRelationship',
    'RelationshipType',
    # Display roles. The role vocabulary is the kiosk's own, so a plugin that
    # reads a reported display type resolves it the same way core does.
    'DashboardDefault',
    'DISPLAY_ROLES',
    'DISPLAY_ROLE_PATHS',
    'normalize_display_role',
    # Response + pagination helpers
    'success_response',
    'error_response',
    'DateValueError',
    'QueryArgError',
    'BadgeShapeError',
    'badge_kind',
    'normalise_badge',
    'intarg',
    'intargs',
    'parse_date',
    'parse_date_fields',
    'paginated_response',
    'ErrorCodes',
    'get_pagination_params',
    'paginate_query',
    # Authorization decorators
    'require_permission',
    'require_any_permission',
    'caller_holds',
    'caller_is_role',
    'require_role',
    'service_token_authorized',
    'authorized_service_token',
    'mint_service_token',
    'list_service_tokens',
    'revoke_service_tokens',
    'collector_authorized',
    'application_package_location',
    'script_body_location',
    'SupportTeam',
    # Serving uploads
    'send_upload',
    'harden_upload_response',
    'client_ip',
    'strip_port',
    # Import-mode helpers
    'apply_import_timestamps',
    'import_mode_active',
    'parse_import_datetime',
    # Legacy employee directory
    'employee_connection',
    'send_email',
    'send_alert',
    'send_webhook',
    'User',
    'Role',
    # CMMC USB check-in/out database
    'cmmc_usb_connection',
]