From 0b928a0ea33090c94131f0fa09c0d33d1d040f0e Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 29 Oct 2026 12:21:11 -0300 Subject: [PATCH] SIENTIAPDE-1974 SIENTIAPDE-1974-connectors-pi-web-api: Implement OPC UA client certificate handling and update connector security model. Connectors now define security using security.mode and security.policy. Client certificate details (MinIO paths) are fetched from the opc_certificates collection via a MongoDB $lookup and included in the OPC slot. A warning is issued if a connector's client_certificate_id does not resolve to an active client certificate. Documentation (README, CLAUDE.md) and tests have been updated. --- CLAUDE.md | 2 +- README.md | 163 ++++++++++++++++-- e2e/helpers.py | 8 +- .../orchestrator_conflict.json | 2 +- .../orchestrator_create_only.json | 2 +- .../orchestrator_happy_path.json | 2 +- e2e/scenario_inputs/orchestrator_noop.json | 2 +- .../orchestrator_tag_database.json | 16 +- .../orchestrator_ttl_index.json | 2 +- .../orchestrator_update_only.json | 2 +- e2e/test_orchestrator_main_workflow.py | 18 +- orchestrator/activities/formatters.py | 35 +++- orchestrator/utils/opc_slots.py | 54 ++++-- orchestrator/utils/tag_database.py | 72 +++++++- orchestrator/workflows/orchestrator.py | 40 ++++- .../activities/test_formatters.py | 61 +++++-- .../utils/tag_database_fixtures.py | 31 +++- tests/orchestrator/utils/test_opc_slots.py | 104 +++++++++-- tests/orchestrator/utils/test_tag_database.py | 67 ++++++- 19 files changed, 595 insertions(+), 88 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 390409b..4b6fb8c 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -47,7 +47,7 @@ CI quality gate (`.github/workflows/quality-gate.yml`) calls the shared reusable - **`orchestrator/worker/worker.py`** — entrypoint (`python -m orchestrator.worker.worker`). Builds a single `Activities` instance, connects to Temporal, starts Prometheus (`HTTP_METRICS_PORT`) and Temporal SDK metrics (`HTTP_SDK_METRICS_PORT`) servers, then runs three `Worker`s concurrently via `asyncio.gather`, each on its own task queue: `orchestrator-queue`, `alerts-queue`, `reports-queue`. `prepare_worker.py` derives the queue name from the workflow class name (`camel_to_snake`) and reads per-queue concurrency/poller tuning from env vars prefixed with the workflow name (e.g. `ORCHESTRATOR_MAX_CONCURRENT_ACTIVITIES`). - **`orchestrator/workflows/`** — Temporal workflow definitions: `orchestrator.py` (pipeline/slot/schedule orchestration), `alerts.py` (real-time ERROR notifications), `reports.py` (scheduled all-level reports). `alerts.py` and `reports.py` both delegate to the same two subworkflows in `workflows/subworkflows/`: `load_notification_package.py` (load notifications + receiver configs from Mongo, incremental via Redis timestamp) and `process_notifications.py` (build HTML email, send, log to Postgres). The difference between alerts and reports is which filter activity they call (`filter_notification_alerts` vs `filter_notification_reports`) and their base data filter (`level: ERROR` by default vs all levels) — `load_notification_package.py`'s `_compute_query_time_filter` can widen either beyond that default to the union of active receiver groups' own `levels` (SIENTIAPDE-2011); the ERROR-only default only holds when no active group narrows further. Receiver groups also support a `model_ids` allow-list (SIENTIAPDE-2149), independent of the pre-existing `ignore_models_id` deny-list — both apply together (AND) when a group configures both, and `model_ids` follows the same union-at-query-time / precise-per-group-slice split as `levels`. `load_latest_data`'s Mongo query is capped at `NOTIFICATION_QUEUE_BATCH_SIZE` documents per run (default `0`, meaning unlimited — unchanged from pre-SIENTIAPDE-2013 behavior unless a deployment opts in), sorted by `timestamp` ascending so a capped run always returns the oldest pending batch first (SIENTIAPDE-2013) — a hard cap with no drain loop, since the incremental Redis cursor picks up any remainder on the next run. - **`orchestrator/activities/activities.py`** — `Activities` is a single class combining every activity mixin via multiple inheritance: `TemporalManager, SlotManager, Formatters, MongoDB, Email, Postgres` (Postgres comes from `sientia_do.temporal.activities.postgres`, the external `sientia-dataops-library` dependency). One instance is constructed in `worker.py` and its bound methods are handed to each `Worker` as the activity list — activities are plain methods on this shared object, not separate classes. Each mixin's `__init__` takes its own config dict plus shared `logger`, `notification_handler`, `metrics_controller`. `Activities.shutdown()` closes every mixin's connections. -- **`orchestrator/utils/`** — `connectors_config.py` builds each config dict from env vars (Temporal/Redis/MongoDB/Postgres/Email); `email_builder.py` renders Jinja2 templates (`utils/templates/email_template.html`, `general_template.html`); `pipeline_configs.py` transforms pipeline documents into pipeline-type-specific configs (`scouter`, `pi_web_api_scouter`, `predictions_batch`, `minimal_retrain`, `drift`, `simple_metrics`); `tag_database.py` indexes the tag database (`connectors` + `connector_tags`) and resolves tags by `tag_id` (scouter read tag names and the `predictions_batch` write tags → `opc_output_config` keyed by `connector_id`, via `resolve_write_tags`; tag changes reach the schedules through `config_hash`, see below); `opc_slots.py` builds the OPC slot v2 the gateway reads (`gather_read_tags` / `build_tag_config`, contract C1: one entry per address with a `names[]` list; see the README section "OPC slots"). `resolve_tag_names` (in `tag_database.py`) is the single source of tag names for both the scouter's `model_tags` and the slot's `names` (SIENTIAPDE-1974 SC1); `converters.py` handles Temporal schedule frequency parsing. +- **`orchestrator/utils/`** — `connectors_config.py` builds each config dict from env vars (Temporal/Redis/MongoDB/Postgres/Email); `email_builder.py` renders Jinja2 templates (`utils/templates/email_template.html`, `general_template.html`); `pipeline_configs.py` transforms pipeline documents into pipeline-type-specific configs (`scouter`, `pi_web_api_scouter`, `predictions_batch`, `minimal_retrain`, `drift`, `simple_metrics`); `tag_database.py` indexes the tag database (`connectors` + `connector_tags`) and resolves tags by `tag_id` (scouter read tag names and the `predictions_batch` write tags → `opc_output_config` keyed by `connector_id`, via `resolve_write_tags`; tag changes reach the schedules through `config_hash`, see below); `opc_slots.py` builds the OPC slot v2 the gateway reads (`gather_read_tags` / `build_tag_config`, contract C1: one entry per address with a `names[]` list, and per server `security` {mode, policy} plus the `certificate` from `tag_database.resolve_certificate`; see the README section "OPC slots"). The certificate comes from `opc_certificates` through the `$lookup` in `DEFAULT_CONNECTORS_QUERY` (only `active` + `client`); no certificate means the gateway connects without security, and a `client_certificate_id` without one is a `WARNING` (SIENTIAPDE-1974, D5). `resolve_tag_names` (in `tag_database.py`) is the single source of tag names for both the scouter's `model_tags` and the slot's `names` (SIENTIAPDE-1974 SC1); `converters.py` handles Temporal schedule frequency parsing. - **Schedule redeploys** — `orchestrated_schedules` stores, per schedule, the last deploy time (`updated_at`) and the `config_hash` it was deployed with (`formatters.schedule_config_hash`: SHA-256 of the built schedule without `updated_at`). `compare_config_timestamps` redeploys when the pipeline's `updated_at` is newer (manual force) or the hash differs (any content change, including tag database edits/deletes made without bumping `updated_at`). `create_schedules`/`update_schedules` pop `config_hash` before sending the schedule to Temporal and return it in their report for `create_pipelines_timestamps`/`update_pipelines_timestamps`. ### Temporal-specific conventions (all workflow/activity files follow this) diff --git a/README.md b/README.md index d5e0077..c4f2170 100644 --- a/README.md +++ b/README.md @@ -103,7 +103,7 @@ The **Orchestrator** workflow is the main coordination workflow that manages pip 4. **Parallel Operations**: Executes slot deletions, slot updates, schedule deletions, schedule creations, and schedule updates concurrently 5. **Parallel Reports & Timestamps**: Generates orchestration reports and updates MongoDB timestamps for created/updated/deleted pipelines -**Input Parameters**: +**Input Parameters** (minimal: each pipeline must already carry `model.name`; see "Workflow Configuration" for the `$lookup` on `models` that adds it): ```json { "schedule_name": "hourly_orchestration", @@ -125,8 +125,19 @@ Optional: `connectors_query` and `connector_tags_query` override the tag databas "collection": "connectors", "aggregation": [ {"$match": {"type": "opc_ua", "status": "Active"}}, - {"$project": {"_id": 0, "connector_id": 1, "name": 1, "url": 1, "server_uri": 1, - "use_security": 1, "subscription_period_ms": 1}} + {"$lookup": {"from": "opc_certificates", + "let": {"certificate_id": "$security.client_certificate_id"}, + "pipeline": [ + {"$match": {"$expr": {"$eq": ["$certificate_id", "$$certificate_id"]}, + "status": "active", "kind": "client"}}, + {"$project": {"_id": 0, "certificate_id": 1, "application_uri": 1, + "cert.bucket": 1, "cert.object_name": 1, + "private_key.bucket": 1, "private_key.object_name": 1}} + ], + "as": "certificates"}}, + {"$project": {"_id": 0, "connector_id": 1, "name": 1, "url": 1, "subscription_period_ms": 1, + "security.mode": 1, "security.policy": 1, + "security.client_certificate_id": 1, "certificates": 1}} ] }, "connector_tags_query": { @@ -145,13 +156,19 @@ The referenced tags are loaded whatever their category and status: the code uses #### OPC slots (slot v2, contract C1) -`process_slots` writes one slot per active ingestor to Redis (`slot:opc_tags:{n}`). The OPC Gateway reads it (SIENTIAPDE-1974). Each server carries its connection fields, and each tag address carries the list of names it is written under: +`process_slots` writes one slot per active ingestor to Redis (`slot:opc_tags:{n}`). The OPC Gateway reads it (SIENTIAPDE-1974). Each server carries its connection fields, its security and the MinIO objects of its client certificate, and each tag address carries the list of names it is written under: ```json { "kepserver": { "server_id": "1", "name": "kepserver", "url": "opc.tcp://host:49320", - "server_uri": "urn:...", "use_security": true, "subscription_period_ms": 7500.0, + "subscription_period_ms": 7500.0, + "security": {"mode": "SignAndEncrypt", "policy": "Basic256Sha256"}, + "certificate": { + "certificate_id": 1, "application_uri": "urn:sientia:opc-ingestor", + "cert": {"bucket": "opc-certificates", "object_name": "client/sientia-platform/opc-ingestor-cert.pem"}, + "private_key": {"bucket": "opc-certificates", "object_name": "client/sientia-platform/opc-ingestor-key.pem"} + }, "tags": { "ns=2;s=Channel1.Device1.Tag1": { "frequency": 15000, @@ -171,7 +188,8 @@ Each scouter `read_tags` entry references its tag by `tag_id` (`{"tag_id": 7, "f | Field | Source | |---|---| | Address | `connector_tags.address` | -| Server (`server_id`, `name`, `url`, `server_uri`, `use_security`, `subscription_period_ms`) | The tag's `connector_tags.connector_id` in `connectors` (`server_id` = `connector_id`) | +| Server (`server_id`, `name`, `url`, `subscription_period_ms`, `security.mode`, `security.policy`) | The tag's `connector_tags.connector_id` in `connectors` (`server_id` = `connector_id`) | +| `certificate` | The `opc_certificates` document of the connector's `security.client_certificate_id`, when it is `active` and of kind `client` (`connectors_query`'s `$lookup`) | | Name | The read tag's `tag_name`, else the `connector_tags` one | Rules: @@ -180,14 +198,16 @@ Rules: - **`frequency`:** the lowest among the pipelines of the address, as an integer. Every read tag needs one. - **`tag_id`:** required, an integer like in `connector_tags`. A read tag whose `tag_id` is missing, `null` or not an integer (`"7"` included) is left out and notified as invalid. - **`subscription_period_ms`:** the connector's value. A connector without one gets half its lowest tag frequency in the slot. -- **`use_security`:** the connector's value. The certificate comes from MinIO on the gateway, so the slot carries no certificate path. +- **`security`:** the connector's `mode` and `policy`, the source of the OPC security (`use_security` and `server_uri` no longer exist). A connector without them is left out like an incomplete one. +- **`certificate`:** `certificate_id`, `application_uri` (the SAN URI of the certificate, used as the client application URI) and the `bucket`/`object_name` of `cert` and `private_key`; the gateway downloads them from MinIO. The certificate dates, `sha256` and audit fields stay out of the slot. There is no default certificate: `null` means the gateway connects without security, and a server that requires a certificate refuses the connection. A connector whose `client_certificate_id` is set but has no active `client` certificate (missing id, `retired`, another kind or an incomplete document) still goes into the slot with `certificate: null`, and is reported (see below). - **`active_read`:** off by default. It is on when any pipeline sets `read_tags[].active_read.enabled`, with the lowest `cycle_multiple`. - **Server keys:** two different connectors with the same name get distinct slot keys (`name`, then `name:server_id`). - **Slot division:** the addresses are split across the slots, sorted by server and address, and the names of an address always stay in the same slot. - **Unresolved tags:** a read tag that cannot be resolved is left out of the slot, and the rest of the slot is still written. It is notified once per run: - `ORCHESTRATOR_TAG_NOT_FOUND_DURING_SLOT_CONFIGURATION`: the `tag_id` is not an active Read tag, or its document lacks `connector_id`, `address` or `tag_name`; - - `ORCHESTRATOR_CONNECTOR_NOT_FOUND_DURING_SLOT_CONFIGURATION`: the connector is missing or inactive, or lacks `name`, `url` or `server_uri`; + - `ORCHESTRATOR_CONNECTOR_NOT_FOUND_DURING_SLOT_CONFIGURATION`: the connector is missing or inactive, or lacks `name`, `url`, `security.mode` or `security.policy`; - `ORCHESTRATOR_INVALID_READ_TAG_DURING_SLOT_CONFIGURATION`: the read tag has no integer `tag_id`, no integer `frequency`, or an enabled `active_read` without an integer `cycle_multiple`; or its pipeline lacks `schedule_name`, `model_id` or `model.name` (the value is then the `schedule_name`). + - `ORCHESTRATOR_CERTIFICATE_NOT_FOUND_DURING_SLOT_CONFIGURATION` (`WARNING`): the connector's `client_certificate_id` has no active `client` certificate, so its tags go into the slot without a certificate. Most likely a registration error. - **Only changed slots are written.** The slot is built in a stable order, so unchanged pipelines produce an identical slot, and the gateway does not re-subscribe. #### Write tags (predictions_batch, through the tag database) @@ -668,6 +688,112 @@ python -m orchestrator.worker.worker | `LOG_LEVEL` | Application logging level | `INFO` | No | | `KAFKA_BOOTSTRAP_SERVERS` | Kafka bootstrap servers | - | No | +### MongoDB Collections + +| Collection | Read / written by | Purpose | +|---|---|---| +| `pipelines` | Orchestrator (`pipelines_query`) | One document per Temporal schedule (`schedule_name`, `workflow_type`, `model_id`, `active`, `frequency`, ...) | +| `models` | Orchestrator (`$lookup` in `pipelines_query`) | Gives each pipeline its `model` (`name`, `active`, `model_config`) | +| `connectors` | Orchestrator (`connectors_query`) | Tag database: the OPC UA servers (`connector_id`, `type`, `name`, `url`, `security`, `subscription_period_ms`, `status`) | +| `connector_tags` | Orchestrator (`connector_tags_query`) | Tag database: the tags (`tag_id`, `connector_id`, `address`, `tag_name`, `category`, `data_type`, `status`) | +| `opc_certificates` | Orchestrator (`$lookup` in `connectors_query`) | The client certificate of each connector (`security.client_certificate_id` → `certificate_id`), only when `active` and of kind `client`: `application_uri` and the MinIO objects of `cert` and `private_key` go into the slot | +| `orchestrated_schedules` | Orchestrator (read and written) | Per deployed schedule: `schedule_name`, `namespace`, `updated_at` (last deploy) and `config_hash` | +| `raw_{schedule_name}` | Orchestrator (created) | One per scouter `topic`, where the OPC Gateway writes its tags. Created with a TTL index on `inserted_at` (`MONGODB_TTL_INDEX_HOURS`) | +| `notification_queue` | Alerts / Reports (read) | Notifications (`notification_id`, `trigger`, `level`, `model_id`, `timestamp`, ...), loaded incrementally by `timestamp` | +| `receiver_groups` | Alerts / Reports (read, `active: true`) | Who receives what (see below) | + +#### Tags in a pipeline + +Pipelines reference tags only by `tag_id` (an integer, as in `connector_tags`). Servers, addresses and data types come from the tag database, never from the pipeline. + +Scouter (`scouter` namespace), `workflow_type: "scouter"`: + +```json +{ + "schedule_name": "scouter-linreg", + "workflow_type": "scouter", + "model_id": "1", + "active": true, + "frequency": "1m", + "read_tags": [ + { + "tag_id": 7, + "frequency": 15000, + "tag_name": "Pressure", + "aggr_func": "avg", + "data_range": [0, 100], + "active_read": {"enabled": true, "cycle_multiple": 2} + } + ] +} +``` + +| Field | Required | Meaning | +|---|---|---| +| `tag_id` | Yes | An active `Read` tag of `connector_tags` | +| `frequency` | Yes | Read period in ms, an integer. The slot uses the lowest among the pipelines of the address | +| `tag_name` | No | Overrides the `connector_tags` name: the name the gateway writes and the scouter looks for | +| `aggr_func` | No | Default `lts` | +| `data_range` | No | Default `[-100, 100]` | +| `active_read` | No | Off by default. When `enabled`, `cycle_multiple` must be an integer (default `1`) | + +Laborious (`laborious` namespace): only `predictions_batch` and `xgboost_predictions_batch` reference tags; `minimal_retrain`, `drift` and `simple_metrics` have none. + +```json +{ + "schedule_name": "pred-linreg", + "workflow_type": "predictions_batch", + "model_id": "1", + "active": true, + "frequency": "1m", + "query": "SELECT ...", + "write_tags": [{"tag_id": 101}, {"tag_id": 102}] +} +``` + +Each `write_tags` entry is only a `tag_id`, of an active `Write` (prediction) or `Prediction Confidence` tag. The Orchestrator builds `opc_output_config` from them (see "Write tags" above) and overwrites any `opc_output_config` stored in the pipeline. + +The `connector_tags` documents those pipelines reference: + +```json +{"tag_id": 7, "connector_id": "plant", "address": "ns=3;i=1001", "tag_name": "Pressure", "category": "Read", "status": "Active"} +{"tag_id": 101, "connector_id": "plant", "address": "ns=3;i=2001", "tag_name": "PRED-101", "category": "Write", "data_type": "double", "status": "Active"} +{"tag_id": 102, "connector_id": "plant", "address": "ns=3;i=2002", "tag_name": "CONF-101", "category": "Prediction Confidence", "status": "Active"} +``` + +`pi_web_api_scouter` (`scouter` namespace) does not use the tag database: its `read_tags` is an object keyed by tag name, and it is not part of the OPC slot: + +```json +{ + "workflow_type": "pi_web_api_scouter", + "read_tags": {"Pressure": {"webid": "F1DP...", "aggr_func": "lts", "data_range": [0, 100]}}, + "pi_web_api_config": {"endpoint": "https://pi/piwebapi", "period": "*-1d", "max_count": 1, "api_timeout": 30} +} +``` + +#### Receiver groups + +```json +{ + "group_name": "admin_team", + "active": true, + "members": ["admin@example.com"], + "contents": ["core_alerts", "persistent_alerts", "reports"], + "levels": ["ERROR", "WARNING"], + "model_ids": ["1"], + "ignore_models_id": ["2"], + "ignore": ["SOME_NOTIFICATION_ID"] +} +``` + +| Field | Meaning | +|---|---| +| `contents` | What the group gets: `core_alerts` / `persistent_alerts` (Alerts workflow), `reports` (Reports workflow) | +| `levels` | Optional allow-list of notification levels. Absent = every level the workflow loads | +| `model_ids` | Optional allow-list of `model_id`s | +| `ignore_models_id` | Optional deny-list of `model_id`s (applies together with `model_ids`) | +| `ignore` | Optional deny-list of `notification_id`s | + ### Workflow Configuration Temporal input configuration sample: @@ -695,7 +821,7 @@ Temporal input configuration sample: }, { "$addFields": { - "models": { + "model": { "$arrayElemAt": [ "$model_docs", 0 @@ -705,7 +831,7 @@ Temporal input configuration sample: }, { "$match": { - "models.active": True + "model.active": True } }, { @@ -719,8 +845,19 @@ Temporal input configuration sample: "collection": "connectors", "aggregation": [ {"$match": {"type": "opc_ua", "status": "Active"}}, - {"$project": {"_id": 0, "connector_id": 1, "name": 1, "url": 1, "server_uri": 1, - "use_security": 1, "subscription_period_ms": 1}} + {"$lookup": {"from": "opc_certificates", + "let": {"certificate_id": "$security.client_certificate_id"}, + "pipeline": [ + {"$match": {"$expr": {"$eq": ["$certificate_id", "$$certificate_id"]}, + "status": "active", "kind": "client"}}, + {"$project": {"_id": 0, "certificate_id": 1, "application_uri": 1, + "cert.bucket": 1, "cert.object_name": 1, + "private_key.bucket": 1, "private_key.object_name": 1}} + ], + "as": "certificates"}}, + {"$project": {"_id": 0, "connector_id": 1, "name": 1, "url": 1, "subscription_period_ms": 1, + "security.mode": 1, "security.policy": 1, + "security.client_certificate_id": 1, "certificates": 1}} ] }, "connector_tags_query": { @@ -735,6 +872,8 @@ Temporal input configuration sample: `connectors_query` and `connector_tags_query` are optional: the values above are the defaults in the code (see "OPC slots" and "Write tags" above). +The `$lookup` must end in a `model` field (singular): every pipeline type reads `model.name` (and `model.model_config`, when present), so a pipeline without `model` fails to build its schedule. + #### Simple metrics: parallel windows & adaptive frequency One `pipelines` doc per window, all with the same `model_id`. Name them `-simple_metrics-m`; each doc becomes its own Temporal schedule. Thresholds are shared through `model.model_config`. The adaptive state is per doc. diff --git a/e2e/helpers.py b/e2e/helpers.py index 830d1ee..e60070f 100644 --- a/e2e/helpers.py +++ b/e2e/helpers.py @@ -148,14 +148,20 @@ def seed_tag_database( database: str, connectors: list[dict[str, Any]], connector_tags: list[dict[str, Any]], + opc_certificates: list[dict[str, Any]] | None = None, ) -> None: - """Insert tag database documents (connectors, connector_tags) into the test Mongo database.""" + """ + Insert tag database documents (connectors, connector_tags and the opc_certificates + the connectors point to) into the test Mongo database. + """ client = MongoClient(mongo_uri) try: if connectors: client[database]['connectors'].insert_many(connectors) if connector_tags: client[database]['connector_tags'].insert_many(connector_tags) + if opc_certificates: + client[database]['opc_certificates'].insert_many(opc_certificates) finally: client.close() diff --git a/e2e/scenario_inputs/orchestrator_conflict.json b/e2e/scenario_inputs/orchestrator_conflict.json index e206610..fd36eb6 100644 --- a/e2e/scenario_inputs/orchestrator_conflict.json +++ b/e2e/scenario_inputs/orchestrator_conflict.json @@ -23,7 +23,7 @@ ], "connectors": [ {"connector_id": "srv-1", "type": "opc_ua", "name": "opc-1", "url": "opc.tcp://opc-1:4840", - "server_uri": "urn:opc-1", "use_security": false, "status": "Active"} + "security": {"mode": "None", "policy": "None"}, "status": "Active"} ], "connector_tags": [ {"tag_id": 101, "connector_id": "srv-1", "address": "ns=2;s=Tag1", "tag_name": "PRED-1", diff --git a/e2e/scenario_inputs/orchestrator_create_only.json b/e2e/scenario_inputs/orchestrator_create_only.json index d3737ce..2d5d94a 100644 --- a/e2e/scenario_inputs/orchestrator_create_only.json +++ b/e2e/scenario_inputs/orchestrator_create_only.json @@ -23,7 +23,7 @@ ], "connectors": [ {"connector_id": "srv-1", "type": "opc_ua", "name": "opc-1", "url": "opc.tcp://opc-1:4840", - "server_uri": "urn:opc-1", "use_security": false, "status": "Active"} + "security": {"mode": "None", "policy": "None"}, "status": "Active"} ], "connector_tags": [ {"tag_id": 101, "connector_id": "srv-1", "address": "ns=2;s=Tag1", "tag_name": "PRED-1", diff --git a/e2e/scenario_inputs/orchestrator_happy_path.json b/e2e/scenario_inputs/orchestrator_happy_path.json index 6d8c29b..5499f75 100644 --- a/e2e/scenario_inputs/orchestrator_happy_path.json +++ b/e2e/scenario_inputs/orchestrator_happy_path.json @@ -35,7 +35,7 @@ ], "connectors": [ {"connector_id": "srv-1", "type": "opc_ua", "name": "opc-1", "url": "opc.tcp://opc-1:4840", - "server_uri": "urn:opc-1", "use_security": false, "status": "Active"} + "security": {"mode": "None", "policy": "None"}, "status": "Active"} ], "connector_tags": [ {"tag_id": 101, "connector_id": "srv-1", "address": "ns=2;s=Tag1", "tag_name": "PRED-1", diff --git a/e2e/scenario_inputs/orchestrator_noop.json b/e2e/scenario_inputs/orchestrator_noop.json index f1d017e..d56ae25 100644 --- a/e2e/scenario_inputs/orchestrator_noop.json +++ b/e2e/scenario_inputs/orchestrator_noop.json @@ -23,7 +23,7 @@ ], "connectors": [ {"connector_id": "srv-1", "type": "opc_ua", "name": "opc-1", "url": "opc.tcp://opc-1:4840", - "server_uri": "urn:opc-1", "use_security": false, "status": "Active"} + "security": {"mode": "None", "policy": "None"}, "status": "Active"} ], "connector_tags": [ {"tag_id": 101, "connector_id": "srv-1", "address": "ns=2;s=Tag1", "tag_name": "PRED-1", diff --git a/e2e/scenario_inputs/orchestrator_tag_database.json b/e2e/scenario_inputs/orchestrator_tag_database.json index a4ea1ea..cab7da2 100644 --- a/e2e/scenario_inputs/orchestrator_tag_database.json +++ b/e2e/scenario_inputs/orchestrator_tag_database.json @@ -12,13 +12,25 @@ "type": "opc_ua", "name": "plant", "url": "opc.tcp://plant:4840", - "server_uri": "urn:plant", - "use_security": false, "subscription_period_ms": 500, + "security": {"mode": "SignAndEncrypt", "policy": "Basic256Sha256", "client_certificate_id": 1}, "status": "Active", "updated_at": "@now-1d" } ], + "opc_certificates": [ + {"certificate_id": 1, "kind": "client", "name": "sientia-platform", "status": "active", + "application_uri": "urn:sientia:opc-ingestor", + "cert": {"bucket": "opc-certificates", "object_name": "client/sientia-platform/cert.pem", "sha256": null}, + "private_key": {"bucket": "opc-certificates", "object_name": "client/sientia-platform/key.pem", "sha256": null}, + "thumbprint_sha1": null, "not_before": "@now-1d", "not_after": "@now+365d", + "created_at": "@now-1d", "updated_at": "@now-1d", "created_by": "e2e", "updated_by": "e2e"}, + {"certificate_id": 2, "kind": "client", "name": "retired", "status": "retired", + "application_uri": "urn:sientia:old", + "cert": {"bucket": "opc-certificates", "object_name": "client/retired/cert.pem"}, + "private_key": {"bucket": "opc-certificates", "object_name": "client/retired/key.pem"}, + "created_at": "@now-1d", "updated_at": "@now-1d", "created_by": "e2e", "updated_by": "e2e"} + ], "connector_tags": [ {"tag_id": 7, "connector_id": "plant", "address": "ns=3;i=1001", "tag_name": "Pressure", "category": "Read", "status": "Active", "updated_at": "@now-1d"}, {"tag_id": 101, "connector_id": "plant", "address": "ns=3;i=2001", "tag_name": "PRED-101", "category": "Write", "data_type": "double", "status": "Active", "updated_at": "@now-1d"}, diff --git a/e2e/scenario_inputs/orchestrator_ttl_index.json b/e2e/scenario_inputs/orchestrator_ttl_index.json index 940156c..d2f6318 100644 --- a/e2e/scenario_inputs/orchestrator_ttl_index.json +++ b/e2e/scenario_inputs/orchestrator_ttl_index.json @@ -24,7 +24,7 @@ ], "connectors": [ {"connector_id": "opc-1", "type": "opc_ua", "name": "opc-1", "url": "opc.tcp://localhost:4840", - "server_uri": "urn:opcfoundation:UA:DemoServer", "use_security": false, "status": "Active"} + "security": {"mode": "None", "policy": "None"}, "status": "Active"} ], "connector_tags": [ {"tag_id": 1, "connector_id": "opc-1", "address": "ns=2;s=Read1", "tag_name": "Read1", diff --git a/e2e/scenario_inputs/orchestrator_update_only.json b/e2e/scenario_inputs/orchestrator_update_only.json index ca3261d..65b4f16 100644 --- a/e2e/scenario_inputs/orchestrator_update_only.json +++ b/e2e/scenario_inputs/orchestrator_update_only.json @@ -23,7 +23,7 @@ ], "connectors": [ {"connector_id": "srv-1", "type": "opc_ua", "name": "opc-1", "url": "opc.tcp://opc-1:4840", - "server_uri": "urn:opc-1", "use_security": false, "status": "Active"} + "security": {"mode": "None", "policy": "None"}, "status": "Active"} ], "connector_tags": [ {"tag_id": 101, "connector_id": "srv-1", "address": "ns=2;s=Tag1", "tag_name": "PRED-1", diff --git a/e2e/test_orchestrator_main_workflow.py b/e2e/test_orchestrator_main_workflow.py index bd20812..fe87e9a 100644 --- a/e2e/test_orchestrator_main_workflow.py +++ b/e2e/test_orchestrator_main_workflow.py @@ -55,6 +55,7 @@ def _apply_orchestrator_seeds( E2E_DATABASE, scenario.get('connectors', []), scenario.get('connector_tags', []), + scenario.get('opc_certificates', []), ) seed_active_ingestors(redis_client, scenario.get('active_ingestors', [])) if scenario.get('orchestrated_schedules'): @@ -534,9 +535,22 @@ async def test_scenario_1_9_3_opc_slot_from_tag_database( 'server_id': 'plant', 'name': 'plant', 'url': 'opc.tcp://plant:4840', - 'server_uri': 'urn:plant', - 'use_security': False, 'subscription_period_ms': 500, + # The active client certificate of client_certificate_id, without its + # dates, sha256 or audit fields (D5) + 'security': {'mode': 'SignAndEncrypt', 'policy': 'Basic256Sha256'}, + 'certificate': { + 'certificate_id': 1, + 'application_uri': 'urn:sientia:opc-ingestor', + 'cert': { + 'bucket': 'opc-certificates', + 'object_name': 'client/sientia-platform/cert.pem', + }, + 'private_key': { + 'bucket': 'opc-certificates', + 'object_name': 'client/sientia-platform/key.pem', + }, + }, 'tags': { 'ns=3;i=1001': { 'frequency': 1000, diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index fdf4d63..0d7909b 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -43,23 +43,35 @@ topic_separator = '\n ========== \n' # Schedule fields left out of config_hash: updated_at is compared on its own CONFIG_HASH_EXCLUDED_FIELDS = ('updated_at', 'config_hash') -# Read tags left out of the slots, by gather_read_tags issue kind +# Read tags left out of the slots, by gather_read_tags issue kind: +# (kind, notification_id, message, level) SLOT_ISSUES = ( ( 'tags', 'ORCHESTRATOR_TAG_NOT_FOUND_DURING_SLOT_CONFIGURATION', 'Tags {} not found among the active, complete Read tags of connector_tags', + NotificationLevel.ERROR, ), ( 'connectors', 'ORCHESTRATOR_CONNECTOR_NOT_FOUND_DURING_SLOT_CONFIGURATION', 'Connectors {} not found, not active or incomplete in connectors', + NotificationLevel.ERROR, ), ( 'invalid', 'ORCHESTRATOR_INVALID_READ_TAG_DURING_SLOT_CONFIGURATION', 'Read tags {} have no integer tag_id, no integer frequency or an active_read without an ' 'integer cycle_multiple; or their pipeline lacks schedule_name, model_id or model.name', + NotificationLevel.ERROR, + ), + ( + 'certificates', + 'ORCHESTRATOR_CERTIFICATE_NOT_FOUND_DURING_SLOT_CONFIGURATION', + 'Connectors {} have a security.client_certificate_id without an active client ' + 'certificate in opc_certificates: they go into the slot without a certificate, so ' + 'without security', + NotificationLevel.WARNING, ), ) @@ -71,16 +83,19 @@ WRITE_TAG_ISSUES = ( 'ORCHESTRATOR_WRITE_TAG_NOT_FOUND_DURING_SCHEDULE_CONFIGURATION', 'Write tags {} not found among the active, complete Write and Prediction Confidence ' 'tags of connector_tags', + NotificationLevel.ERROR, ), ( 'connectors', 'ORCHESTRATOR_WRITE_TAG_CONNECTOR_NOT_FOUND_DURING_SCHEDULE_CONFIGURATION', 'Connectors {} of write tags not found, not active or incomplete in connectors', + NotificationLevel.ERROR, ), ( 'invalid', 'ORCHESTRATOR_INVALID_WRITE_TAG_DURING_SCHEDULE_CONFIGURATION', 'Write tags {} have no integer tag_id or a data_type the OPC Gateway does not accept', + NotificationLevel.ERROR, ), ) @@ -330,14 +345,17 @@ class Formatters(SientiaMonitoring): The division is by address: all the names of an address stay in the same slot (SIENTIAPDE-1974, O5). Read tags that cannot be resolved are left out and notified once per run: a read tag without tag_id or a usable frequency, a tag_id - missing from connector_tags, or a connector missing or inactive. + missing from connector_tags, or a connector missing or inactive. A connector whose + client_certificate_id has no active client certificate goes into the slot without + a certificate, with a warning (D5). Args: - input_data (dict[str, Any]): The input data containing the schedules to process. - pipelines (list[dict[str, Any]]): The schedules to process. - connectors (list[dict[str, Any]], optional): Active OPC UA connectors of - the tag database, the only source of the OPC servers. + the tag database, the only source of the OPC servers, each with the + certificates of its client_certificate_id (connectors_query). - connector_tags (list[dict[str, Any]], optional): The connector_tags the pipelines reference; only the active Read tags are used. - active_ingestors (list[str]): The active ingestors to divide into slots. @@ -374,7 +392,7 @@ class Formatters(SientiaMonitoring): def _notify_issues( self, issues: dict[str, set[str]], - kinds: tuple[tuple[str, str, str], ...], + kinds: tuple[tuple[str, str, str, str], ...], metadata: dict[str, Any], ) -> None: """ @@ -382,18 +400,19 @@ class Formatters(SientiaMonitoring): Args: issues (dict[str, set[str]]): What was left out, by kind - kinds (tuple[tuple[str, str, str], ...]): (kind, notification_id, message with - one {} for the joined values), e.g. SLOT_ISSUES or WRITE_TAG_ISSUES + kinds (tuple[tuple[str, str, str, str], ...]): (kind, notification_id, message + with one {} for the joined values, level), e.g. SLOT_ISSUES or + WRITE_TAG_ISSUES metadata (dict[str, Any]): Metadata for the notifications """ - for kind, notification_id, message in kinds: + for kind, notification_id, message, level in kinds: if issues.get(kind): self.send_notification( metadata=metadata, notification_id=notification_id, message=message.format(', '.join(sorted(issues[kind]))), block='orchestrator', - level=NotificationLevel.ERROR, + level=level, ) @activity.defn(name='format_schedule_config') diff --git a/orchestrator/utils/opc_slots.py b/orchestrator/utils/opc_slots.py index f6c8113..0bc6814 100644 --- a/orchestrator/utils/opc_slots.py +++ b/orchestrator/utils/opc_slots.py @@ -2,18 +2,35 @@ from typing import Any, cast -from orchestrator.utils.tag_database import is_complete, resolve_tag_names, tag_id_of +from orchestrator.utils.tag_database import ( + client_certificate_id, + is_complete, + resolve_certificate, + resolve_tag_names, + tag_id_of, +) -def _connector_server(connector: dict[str, Any]) -> dict[str, Any]: - """Slot server fields of a tag database connector (connectors collection).""" +def _connector_server(connector: dict[str, Any], issues: dict[str, set[str]]) -> dict[str, Any]: + """ + Slot server fields of a tag database connector (connectors collection). + + A connector whose client_certificate_id has no active client certificate still goes + into the slot, without a certificate, and is recorded in issues['certificates'] (D5). + """ + certificate = resolve_certificate(connector) + if certificate is None and client_certificate_id(connector) is not None: + issues['certificates'].add(str(connector['connector_id'])) return { 'server_id': str(connector['connector_id']), 'name': connector['name'], 'url': connector['url'], - 'server_uri': connector['server_uri'], - 'use_security': bool(connector.get('use_security', False)), 'subscription_period_ms': connector.get('subscription_period_ms'), + 'security': { + 'mode': connector['security']['mode'], + 'policy': connector['security']['policy'], + }, + 'certificate': certificate, } @@ -35,7 +52,8 @@ def _locate_read_tag( Server and address of a read tag, or None when they cannot be resolved. The address and connector come from connector_tags and the server from connectors - (RF3). What cannot be resolved is recorded in issues ('tags' or 'connectors'). + (RF3). What cannot be resolved is recorded in issues ('tags' or 'connectors'), and a + server without its registered certificate in issues['certificates']. """ connector_tag = tags_by_id.get(tag_id) if connector_tag is None: @@ -46,7 +64,7 @@ def _locate_read_tag( if connector is None: issues['connectors'].add(connector_id) return None - return _connector_server(connector), connector_tag['address'] + return _connector_server(connector, issues), connector_tag['address'] def _cycle_multiple(active_read: Any) -> int | None: @@ -195,9 +213,10 @@ def gather_read_tags( Returns: tuple[list[dict[str, Any]], dict[str, set[str]]]: - One entry per (server, address), sorted by server_id then address, with: - - server (dict): server_id, name, url, server_uri, use_security, - subscription_period_ms (None = derive from the frequencies) and key - (the server's unique slot key) + - server (dict): server_id, name, url, subscription_period_ms (None = + derive from the frequencies), security ({mode, policy}), certificate + (resolve_certificate; None = connect without security) and key (the + server's unique slot key) - address (str): The OPC node address - frequency (int): The lowest frequency among the pipelines (RF6) - active_read (dict): {'enabled': False} unless a pipeline enables it (D3) @@ -206,12 +225,14 @@ def gather_read_tags( - What was left out: 'tags' (tag_id), 'connectors' (connector_id) and 'invalid' (schedule_name/read tag without an int tag_id, a usable frequency or cycle_multiple, or the schedule_name of a pipeline without schedule_name, - model_id or model.name) + model_id or model.name); and 'certificates': connector_ids that went into the + slot without the certificate their client_certificate_id points to """ issues: dict[str, set[str]] = { 'tags': set(), 'connectors': set(), 'invalid': set(), + 'certificates': set(), } servers: dict[str, dict[str, Any]] = {} entries: dict[tuple[str, str], dict[str, Any]] = {} @@ -238,15 +259,16 @@ def gather_read_tags( def build_tag_config(entries: list[dict[str, Any]]) -> dict[str, Any]: """ - Build one slot in the slot v2 format (contract C1) from address entries. The slot - carries no certificate: the gateway takes it from MinIO (SIENTIAPDE-1974, O4). + Build one slot in the slot v2 format (contract C1) from address entries. Each server + carries its security and the MinIO objects of its client certificate, which the + gateway downloads (SIENTIAPDE-1974); certificate None means no security (D5). Args: entries (list[dict[str, Any]]): Address entries from gather_read_tags for this slot Returns: dict[str, Any]: The slot keyed by server key, each server with server_id, name, - url, server_uri, use_security, subscription_period_ms and tags. Each tag (keyed + url, subscription_period_ms, security, certificate and tags. Each tag (keyed by address) has frequency, active_read and names. A connector without a subscription_period_ms gets half its lowest tag frequency in this slot. """ @@ -259,9 +281,9 @@ def build_tag_config(entries: list[dict[str, Any]]) -> dict[str, Any]: 'server_id': server['server_id'], 'name': server['name'], 'url': server['url'], - 'server_uri': server['server_uri'], - 'use_security': server['use_security'], 'subscription_period_ms': server['subscription_period_ms'], + 'security': server['security'], + 'certificate': server['certificate'], 'tags': {}, } diff --git a/orchestrator/utils/tag_database.py b/orchestrator/utils/tag_database.py index 8aeb08c..9b22126 100644 --- a/orchestrator/utils/tag_database.py +++ b/orchestrator/utils/tag_database.py @@ -11,7 +11,17 @@ from typing import Any CONNECTOR_TAG_FIELDS = ('tag_id', 'connector_id', 'address', 'tag_name') -CONNECTOR_FIELDS = ('connector_id', 'name', 'url', 'server_uri') +CONNECTOR_FIELDS = ('connector_id', 'name', 'url') + + +# security fields of an OPC UA connector: the source of its OPC security (D4) +SECURITY_FIELDS = ('mode', 'policy') + + +# Fields of a client certificate the slot carries, and of each object it names in MinIO +CERTIFICATE_FIELDS = ('certificate_id', 'application_uri') +CERTIFICATE_OBJECTS = ('cert', 'private_key') +STORED_OBJECT_FIELDS = ('bucket', 'object_name') # connector_tags categories each use accepts @@ -146,8 +156,8 @@ def index_connectors(connectors: list[dict[str, Any]]) -> dict[str, dict[str, An """ Index the tag database's connectors by connector_id (as text). - Documents that are not active or lack connector_id, name, url or server_uri are - left out. + Documents that are not active or lack connector_id, name, url, security.mode or + security.policy are left out. Args: connectors (list[dict[str, Any]]): connectors documents (SIENTIAPDE-1569) @@ -158,10 +168,64 @@ def index_connectors(connectors: list[dict[str, Any]]) -> dict[str, dict[str, An return { str(connector['connector_id']): connector for connector in connectors - if is_complete(connector, CONNECTOR_FIELDS) and _is_active(connector) + if is_complete(connector, CONNECTOR_FIELDS) + and _has_security(connector) + and _is_active(connector) } +def _has_security(connector: dict[str, Any]) -> bool: + security = connector.get('security') + return isinstance(security, dict) and is_complete(security, SECURITY_FIELDS) + + +def client_certificate_id(connector: dict[str, Any]) -> Any: + """The connector's security.client_certificate_id (None when it has none).""" + security = connector.get('security') + return security.get('client_certificate_id') if isinstance(security, dict) else None + + +def _is_usable_certificate(certificate: Any) -> bool: + return ( + isinstance(certificate, dict) + and is_complete(certificate, CERTIFICATE_FIELDS) + and all( + isinstance(certificate.get(name), dict) + and is_complete(certificate[name], STORED_OBJECT_FIELDS) + for name in CERTIFICATE_OBJECTS + ) + ) + + +def resolve_certificate(connector: dict[str, Any]) -> dict[str, Any] | None: + """ + The client certificate a connector goes into the slot with, or None (D5). + + connectors_query's $lookup puts in `certificates` the opc_certificates document the + connector's security.client_certificate_id points to, only when it is active and of + kind client. None means no certificate: the gateway connects without security. A + connector with a client_certificate_id and no certificate is most likely a + registration error, reported by the caller (see client_certificate_id). + + Args: + connector (dict[str, Any]): A connectors document from connectors_query + + Returns: + dict[str, Any] | None: certificate_id, application_uri, and cert and private_key + as {bucket, object_name}; None without a complete certificate + """ + for certificate in connector.get('certificates') or []: + if _is_usable_certificate(certificate): + return { + **{field: certificate[field] for field in CERTIFICATE_FIELDS}, + **{ + name: {field: certificate[name][field] for field in STORED_OBJECT_FIELDS} + for name in CERTIFICATE_OBJECTS + }, + } + return None + + def resolve_tag_names( read_tags: list[dict[str, Any]], tags_by_id: dict[int, dict[str, Any]] ) -> list[dict[str, Any]]: diff --git a/orchestrator/workflows/orchestrator.py b/orchestrator/workflows/orchestrator.py index fa60a69..0b5f839 100644 --- a/orchestrator/workflows/orchestrator.py +++ b/orchestrator/workflows/orchestrator.py @@ -15,6 +15,7 @@ with workflow.unsafe.imports_passed_through(): # MongoDB aggregation stages MATCH = '$match' PROJECT = '$project' +LOOKUP = '$lookup' # Tag database (SIENTIAPDE-1569) read by the slot v2 and the schedules (SIENTIAPDE-1974). # The workflow input may override these aggregations (connectors_query / @@ -22,20 +23,53 @@ PROJECT = '$project' # connector_tags_query. Every category and status of the referenced tags is loaded: the # slot uses the active Read tags, predictions_batch the active Write / Prediction # Confidence ones (status and category are filtered in the code). Only the fields used -# are projected: other attributes may hold BSON types a Temporal payload cannot carry. +# are projected: other attributes may hold BSON types a Temporal payload cannot carry +# (so the certificate dates are left out). +# Each connector brings, in `certificates`, the opc_certificates document its +# security.client_certificate_id points to, only when it is active and of kind client +# (D5); resolve_certificate turns it into the slot's certificate. DEFAULT_CONNECTORS_QUERY = { 'collection': 'connectors', 'aggregation': [ {MATCH: {'type': 'opc_ua', 'status': 'Active'}}, + { + LOOKUP: { + 'from': 'opc_certificates', + 'let': {'certificate_id': '$security.client_certificate_id'}, + 'pipeline': [ + { + MATCH: { + '$expr': {'$eq': ['$certificate_id', '$$certificate_id']}, + 'status': 'active', + 'kind': 'client', + } + }, + { + PROJECT: { + '_id': 0, + 'certificate_id': 1, + 'application_uri': 1, + 'cert.bucket': 1, + 'cert.object_name': 1, + 'private_key.bucket': 1, + 'private_key.object_name': 1, + } + }, + ], + 'as': 'certificates', + } + }, { PROJECT: { '_id': 0, 'connector_id': 1, 'name': 1, 'url': 1, - 'server_uri': 1, - 'use_security': 1, 'subscription_period_ms': 1, + 'security.mode': 1, + 'security.policy': 1, + 'security.client_certificate_id': 1, + 'certificates': 1, } }, ], diff --git a/tests/orchestrator/activities/test_formatters.py b/tests/orchestrator/activities/test_formatters.py index 60b96fe..d479379 100644 --- a/tests/orchestrator/activities/test_formatters.py +++ b/tests/orchestrator/activities/test_formatters.py @@ -7,6 +7,12 @@ from sientia_do.notifications.models import NotificationLevel from orchestrator.activities.formatters import Formatters, schedule_config_hash from orchestrator.utils.pipeline_configs import predictions_batch +from tests.orchestrator.utils.tag_database_fixtures import ( + CERTIFICATE, + NO_SECURITY, + SECURE, + secure_connector_fields, +) @fixture @@ -237,7 +243,7 @@ slot_connectors = [ 'connector_id': 'kepserver', 'name': 'kepserver', 'url': 'opc.tcp://kep:49320', - 'server_uri': 'urn:kep', + 'security': NO_SECURITY, 'status': 'Active', }, ] @@ -293,8 +299,8 @@ def test_process_slots(formatters): 'server_id': 'kepserver', 'name': 'kepserver', 'url': 'opc.tcp://kep:49320', - 'server_uri': 'urn:kep', - 'use_security': False, + 'security': NO_SECURITY, + 'certificate': None, } assert result == { '1': { @@ -392,10 +398,9 @@ def test_process_slots_with_tag_database(formatters): 'type': 'opc_ua', 'name': 'plant', 'url': 'opc.tcp://plant:4840', - 'server_uri': 'urn:plant', - 'use_security': True, 'subscription_period_ms': 1500, 'status': 'Active', + **secure_connector_fields(), } ], 'connector_tags': [ @@ -414,9 +419,9 @@ def test_process_slots_with_tag_database(formatters): 'server_id': '3', 'name': 'plant', 'url': 'opc.tcp://plant:4840', - 'server_uri': 'urn:plant', - 'use_security': True, 'subscription_period_ms': 1500, + 'security': SECURE, + 'certificate': CERTIFICATE, 'tags': { 'ns=3;i=1001': { 'frequency': 1000, @@ -491,6 +496,42 @@ def test_process_slots_notifies_unresolved_tags_once(formatters): assert formatters.send_notification.call_count == 3 +def test_process_slots_warns_about_a_missing_certificate(formatters): + """A client_certificate_id without an active client certificate: the server goes into + the slot without a certificate, and a warning names the connector (D5)""" + connectors = [ + {**slot_connectors[0], **secure_connector_fields(certificate_id=9, certificates=[])}, + { + **slot_connectors[0], + 'connector_id': 'plant', + 'name': 'plant', + **secure_connector_fields(), + }, + ] + connector_tags = [*slot_connector_tags, _read_tag_doc(7, 'ns=3;i=7', 'P', 'plant')] + pipelines = [ + _scouter_pipeline('sin', 'm1', [{'tag_id': 1, 'frequency': 1000}]), + _scouter_pipeline('cos', 'm2', [{'tag_id': 7, 'frequency': 1000}]), + ] + + result = formatters.process_slots( + _slot_input(pipelines, [], connectors=connectors, connector_tags=connector_tags) + ) + + assert result['1']['kepserver']['certificate'] is None + assert result['1']['kepserver']['security'] == SECURE + assert result['1']['plant']['certificate'] == CERTIFICATE + formatters.send_notification.assert_called_once_with( + metadata=metadata['metadata'], + notification_id='ORCHESTRATOR_CERTIFICATE_NOT_FOUND_DURING_SLOT_CONFIGURATION', + message='Connectors kepserver have a security.client_certificate_id without an active ' + 'client certificate in opc_certificates: they go into the slot without a certificate, ' + 'so without security', + block='orchestrator', + level=NotificationLevel.WARNING, + ) + + def test_process_slots_survives_malformed_tag_database(formatters): """An incomplete connector or tag leaves only its own read tags out of the slot""" pipelines = [ @@ -504,7 +545,7 @@ def test_process_slots_survives_malformed_tag_database(formatters): ], ), ] - connectors = [*slot_connectors, {'connector_id': 3, 'name': 'plant', 'server_uri': 'urn:plant'}] + connectors = [*slot_connectors, {'connector_id': 3, 'name': 'plant', 'security': NO_SECURITY}] connector_tags = [ *slot_connector_tags, {'tag_id': 7, 'connector_id': 3, 'address': 'ns=3;i=1', 'tag_name': 'P'}, @@ -554,7 +595,7 @@ def test_process_schedules_names_scouter_tags_like_the_slot(formatters): 'connector_id': 3, 'name': 'plant', 'url': 'opc.tcp://plant:4840', - 'server_uri': 'urn:plant', + 'security': NO_SECURITY, } ] @@ -1470,7 +1511,7 @@ PLANT_CONNECTOR = { 'connector_id': 'plant', 'name': 'plant', 'url': 'opc.tcp://plant:4840', - 'server_uri': 'urn:plant', + 'security': NO_SECURITY, 'status': 'Active', } diff --git a/tests/orchestrator/utils/tag_database_fixtures.py b/tests/orchestrator/utils/tag_database_fixtures.py index ece49be..80b6303 100644 --- a/tests/orchestrator/utils/tag_database_fixtures.py +++ b/tests/orchestrator/utils/tag_database_fixtures.py @@ -17,6 +17,35 @@ TAGS_BY_ID = { } +# Connector security and client certificates (SIENTIAPDE-1974) + +NO_SECURITY = {'mode': 'None', 'policy': 'None'} +SECURE = {'mode': 'SignAndEncrypt', 'policy': 'Basic256Sha256'} + + +# A client certificate as connectors_query's $lookup brings it (certificates[]) +def certificate_doc(certificate_id=1, name='sientia-platform', **extra): + return { + 'certificate_id': certificate_id, + 'application_uri': 'urn:sientia:opc-ingestor', + 'cert': {'bucket': 'opc-certificates', 'object_name': f'client/{name}/cert.pem'}, + 'private_key': {'bucket': 'opc-certificates', 'object_name': f'client/{name}/key.pem'}, + **extra, + } + + +# The slot's certificate for certificate_doc(): the same fields +CERTIFICATE = certificate_doc() + + +def secure_connector_fields(certificate_id=1, certificates=None): + """security and certificates of a connector with a client certificate.""" + return { + 'security': {**SECURE, 'client_certificate_id': certificate_id}, + 'certificates': [certificate_doc(certificate_id)] if certificates is None else certificates, + } + + # Write tags (predictions_batch write_tags through the tag database) OPC_CONNECTORS = { @@ -24,7 +53,7 @@ OPC_CONNECTORS = { 'connector_id': 'plant', 'name': 'plant', 'url': 'opc.tcp://plant:4840', - 'server_uri': 'urn:plant', + 'security': NO_SECURITY, }, } diff --git a/tests/orchestrator/utils/test_opc_slots.py b/tests/orchestrator/utils/test_opc_slots.py index f9ecc13..f32afbb 100644 --- a/tests/orchestrator/utils/test_opc_slots.py +++ b/tests/orchestrator/utils/test_opc_slots.py @@ -1,12 +1,21 @@ +import pytest + from orchestrator.utils.opc_slots import ( build_tag_config, gather_read_tags, ) from tests.orchestrator.utils.tag_database_fixtures import ( + CERTIFICATE, + NO_SECURITY, + SECURE, TAGS_BY_ID, + certificate_doc, scouter_pipeline_doc, + secure_connector_fields, ) +NO_ISSUES = {'tags': set(), 'connectors': set(), 'invalid': set(), 'certificates': set()} + def _name(schedule_name, tag_name, model_id): return { @@ -26,7 +35,7 @@ def _connector(connector_id, name, **extra): 'connector_id': connector_id, 'name': name, 'url': f'opc.tcp://{name}:4840', - 'server_uri': f'urn:{name}', + 'security': NO_SECURITY, **extra, } @@ -41,8 +50,8 @@ def _db_tag(tag_id, connector_id, address, tag_name): connectors = { - '1': _connector(1, 'kepserver', use_security=False), - '3': _connector(3, 'plant', use_security=True, subscription_period_ms=1500), + '1': _connector(1, 'kepserver'), + '3': _connector(3, 'plant', subscription_period_ms=1500, **secure_connector_fields()), } @@ -77,7 +86,7 @@ def test_gather_read_tags_multi_name(): assert entries[0]['address'] == 'ns=2;s=A' assert entries[0]['frequency'] == 5000 assert entries[0]['names'] == [_name('cos', 'Contador', 'm2'), _name('sin', 'Counter', 'm1')] - assert issues == {'tags': set(), 'connectors': set(), 'invalid': set()} + assert issues == NO_ISSUES def test_gather_read_tags_same_name_in_two_pipelines(): @@ -125,9 +134,9 @@ def test_gather_read_tags_with_tag_database(): 'server_id': '3', 'name': 'plant', 'url': 'opc.tcp://plant:4840', - 'server_uri': 'urn:plant', - 'use_security': True, 'subscription_period_ms': 1500, + 'security': SECURE, + 'certificate': CERTIFICATE, 'key': 'plant', }, 'address': 'ns=3;i=1001', @@ -148,7 +157,7 @@ def test_gather_read_tags_unknown_tag_id(): entries, issues = _gather([scouter_pipeline_doc('sin', 'm1', [_read(404)])]) assert entries == [] - assert issues == {'tags': {'404'}, 'connectors': set(), 'invalid': set()} + assert issues == {**NO_ISSUES, 'tags': {'404'}} def test_gather_read_tags_inactive_connector(): @@ -156,7 +165,7 @@ def test_gather_read_tags_inactive_connector(): entries, issues = _gather([scouter_pipeline_doc('sin', 'm1', [_read(8)])]) assert entries == [] - assert issues == {'tags': set(), 'connectors': {'5'}, 'invalid': set()} + assert issues == {**NO_ISSUES, 'connectors': {'5'}} def test_gather_read_tags_without_tag_id(): @@ -179,11 +188,7 @@ def test_gather_read_tags_without_tag_id(): entries, issues = _gather(pipelines) assert [e['address'] for e in entries] == ['ns=2;s=A'] - assert issues == { - 'tags': set(), - 'connectors': set(), - 'invalid': {'sin/b', 'sin/c', 'sin/d', 'sin/11'}, - } + assert issues == {**NO_ISSUES, 'invalid': {'sin/b', 'sin/c', 'sin/d', 'sin/11'}} def test_gather_read_tags_server_name_clash(): @@ -241,11 +246,11 @@ def test_gather_read_tags_without_read_tags(): pipeline = scouter_pipeline_doc('sin', 'm1', []) del pipeline['read_tags'] - assert _gather([pipeline]) == ([], {'tags': set(), 'connectors': set(), 'invalid': set()}) + assert _gather([pipeline]) == ([], NO_ISSUES) def test_build_tag_config(): - """Slot v2 (contract C1): no certificate paths, subscription period per server""" + """Slot v2 (contract C1): security, certificate and subscription period per server""" pipelines = [ scouter_pipeline_doc('sin', 'm1', [_read(10, 1000), _read(11, '300'), _read(7, 100)]) ] @@ -258,8 +263,9 @@ def test_build_tag_config(): 'server_id': '1', 'name': 'kepserver', 'url': 'opc.tcp://kepserver:4840', - 'server_uri': 'urn:kepserver', - 'use_security': False, + # No client certificate: no security (D5) + 'security': NO_SECURITY, + 'certificate': None, # No subscription_period_ms on the connector: half its lowest tag frequency 'subscription_period_ms': 150, 'tags': { @@ -279,9 +285,9 @@ def test_build_tag_config(): 'server_id': '3', 'name': 'plant', 'url': 'opc.tcp://plant:4840', - 'server_uri': 'urn:plant', - 'use_security': True, 'subscription_period_ms': 1500, + 'security': SECURE, + 'certificate': CERTIFICATE, 'tags': { 'ns=3;i=1001': { 'frequency': 100, @@ -382,3 +388,63 @@ def test_gather_read_tags_skips_incomplete_pipelines(): assert [e['address'] for e in entries] == ['ns=2;s=B'] assert issues['invalid'] == {'no-model', 'no-name', 'no-id', 'None'} + + +def _plant_with(**fields): + return {'1': connectors['1'], '3': {**connectors['3'], **fields}} + + +@pytest.mark.parametrize( + 'fields', + [ + # $lookup found nothing: missing id, inactive or not a client certificate + pytest.param(secure_connector_fields(certificates=[]), id='not-found'), + pytest.param( + secure_connector_fields(certificates=[certificate_doc(application_uri='')]), + id='no-application-uri', + ), + pytest.param( + secure_connector_fields( + certificates=[{**certificate_doc(), 'private_key': {'bucket': 'b'}}] + ), + id='key-without-object', + ), + ], +) +def test_gather_read_tags_registered_certificate_missing(fields): + """An id without a usable certificate: in the slot without a certificate, with a warning (D5)""" + entries, issues = _gather( + [scouter_pipeline_doc('sin', 'm1', [_read(7)])], _plant_with(**fields) + ) + + assert entries[0]['server']['certificate'] is None + assert entries[0]['server']['security'] == SECURE + assert issues == {**NO_ISSUES, 'certificates': {'3'}} + + +@pytest.mark.parametrize('certificate_id', [None, 'absent']) +def test_gather_read_tags_without_certificate_id(certificate_id): + """No client_certificate_id: no certificate and nothing to report (D5)""" + security = {**NO_SECURITY, 'client_certificate_id': certificate_id} + if certificate_id == 'absent': + security = NO_SECURITY + entries, issues = _gather( + [scouter_pipeline_doc('sin', 'm1', [_read(7)])], + _plant_with(security=security, certificates=[]), + ) + + assert entries[0]['server']['certificate'] is None + assert issues == NO_ISSUES + + +def test_gather_read_tags_reports_a_certificate_once_per_connector(): + entries, issues = _gather( + [ + scouter_pipeline_doc('sin', 'm1', [_read(7)]), + scouter_pipeline_doc('cos', 'm2', [_read(7)]), + ], + _plant_with(**secure_connector_fields(certificates=[])), + ) + + assert len(entries) == 1 + assert issues['certificates'] == {'3'} diff --git a/tests/orchestrator/utils/test_tag_database.py b/tests/orchestrator/utils/test_tag_database.py index 48d246f..05ce73e 100644 --- a/tests/orchestrator/utils/test_tag_database.py +++ b/tests/orchestrator/utils/test_tag_database.py @@ -1,19 +1,27 @@ +import pytest + from orchestrator.utils.tag_database import ( WRITE_CATEGORIES, + client_certificate_id, filter_by_tag_ids, index_connector_tags, index_connectors, index_tag_documents, referenced_tag_ids, + resolve_certificate, resolve_tag_names, resolve_write_tags, tag_id_of, ) from tests.orchestrator.utils.tag_database_fixtures import ( + CERTIFICATE, + NO_SECURITY, OPC_CONNECTORS, TAGS_BY_ID, batch_pipeline_doc, + certificate_doc, scouter_pipeline_doc, + secure_connector_fields, write_tag_doc, ) @@ -92,9 +100,23 @@ def test_index_skips_incomplete_documents(): {'tag_id': 9, 'connector_id': 3, 'address': '', 'tag_name': 'no address'}, ] connector_list = [ - {'connector_id': 3, 'name': 'plant', 'url': 'opc.tcp://p:4840', 'server_uri': 'urn:p'}, - {'connector_id': 4, 'name': 'no url', 'server_uri': 'urn:x'}, - {'name': 'no id', 'url': 'opc.tcp://x:4840', 'server_uri': 'urn:x'}, + {'connector_id': 3, 'name': 'plant', 'url': 'opc.tcp://p:4840', 'security': NO_SECURITY}, + {'connector_id': 4, 'name': 'no url', 'security': NO_SECURITY}, + {'name': 'no id', 'url': 'opc.tcp://x:4840', 'security': NO_SECURITY}, + # security.mode and security.policy are the source of the OPC security (D4) + {'connector_id': 5, 'name': 'no security', 'url': 'opc.tcp://x:4840'}, + { + 'connector_id': 6, + 'name': 'no mode', + 'url': 'opc.tcp://x:4840', + 'security': {'policy': 'None'}, + }, + { + 'connector_id': 7, + 'name': 'no policy', + 'url': 'opc.tcp://x:4840', + 'security': {'mode': 'None'}, + }, ] assert list(index_connector_tags(tags)) == [7] @@ -216,3 +238,42 @@ def test_resolve_write_tags_issues(): "{'tag_id': None}", 'ns=2;s=raw', } + + +def test_resolve_certificate(): + """Only the fields the gateway needs: dates and other fields stay out of the slot""" + connector = { + **OPC_CONNECTORS['plant'], + **secure_connector_fields( + certificates=[certificate_doc(thumbprint_sha1='ab', not_after=None)] + ), + } + + assert resolve_certificate(connector) == CERTIFICATE + assert client_certificate_id(connector) == 1 + + +@pytest.mark.parametrize( + 'certificates', + [ + pytest.param(None, id='no-lookup'), + pytest.param([], id='not-found'), + pytest.param([certificate_doc(application_uri=None)], id='no-application-uri'), + pytest.param([{**certificate_doc(), 'cert': None}], id='no-cert'), + pytest.param( + [{**certificate_doc(), 'private_key': {'object_name': 'k'}}], id='key-no-bucket' + ), + pytest.param(['not a document'], id='not-a-document'), + ], +) +def test_resolve_certificate_without_a_usable_certificate(certificates): + connector = {**OPC_CONNECTORS['plant'], **secure_connector_fields()} + connector['certificates'] = certificates + + assert resolve_certificate(connector) is None + + +def test_client_certificate_id_without_one(): + assert client_certificate_id(OPC_CONNECTORS['plant']) is None + assert client_certificate_id({'connector_id': 'x'}) is None + assert client_certificate_id({'security': None}) is None