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

@@ -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)

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,
+      {"$lookup": {"from": "opc_certificates",
-                    "use_security": 1, "subscription_period_ms": 1}}
+                   "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,
+            {"$lookup": {"from": "opc_certificates",
-                          "use_security": 1, "subscription_period_ms": 1}}
+                         "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 `<model>-simple_metrics-<interval_minutes>m`; each doc becomes its own Temporal schedule. Thresholds are shared through `model.model_config`. The adaptive state is per doc.

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()
 

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",

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",

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",

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",

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"},

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",

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",

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,

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
+            kinds (tuple[tuple[str, str, str, str], ...]): (kind, notification_id, message
-                one {} for the joined values), e.g. SLOT_ISSUES or WRITE_TAG_ISSUES
+                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')

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]:
+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)."""
+    """
+    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,
+                - server (dict): server_id, name, url, subscription_period_ms (None =
-                  subscription_period_ms (None = derive from the frequencies) and key
+                  derive from the frequencies), security ({mode, policy}), certificate
-                  (the server's unique slot key)
+                  (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
+    Build one slot in the slot v2 format (contract C1) from address entries. Each server
-    carries no certificate: the gateway takes it from MinIO (SIENTIAPDE-1974, O4).
+    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': {},
             }
 

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
+    Documents that are not active or lack connector_id, name, url, security.mode or
-    left out.
+    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]]:

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,
             }
         },
     ],

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',
+        'security': NO_SECURITY,
-        'use_security': False,
+        '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',
 }
 

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,
     },
 }
 

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),
+    '1': _connector(1, 'kepserver'),
-    '3': _connector(3, 'plant', use_security=True, subscription_period_ms=1500),
+    '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 == {
+    assert issues == {**NO_ISSUES, 'invalid': {'sin/b', 'sin/c', 'sin/d', 'sin/11'}}
-        'tags': set(),
-        'connectors': set(),
-        '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',
+            # No client certificate: no security (D5)
-            'use_security': False,
+            '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'}

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': 3, 'name': 'plant', 'url': 'opc.tcp://p:4840', 'security': NO_SECURITY},
-        {'connector_id': 4, 'name': 'no url', 'server_uri': 'urn:x'},
+        {'connector_id': 4, 'name': 'no url', 'security': NO_SECURITY},
-        {'name': 'no id', 'url': 'opc.tcp://x:4840', 'server_uri': 'urn:x'},
+        {'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