Документация

Аудитория: интегратор, суперадминистратор, инженер внедрения. Закрывает вопрос: «что конкретно сделать и куда вписать, чтобы вместо фикстуры пошли живые данные KSC/1С/журналов — и при этом не сломать гардрейлы?». Предшествующие статьи (обязательно прочесть): 03 — Коннекторы под капотом, 04 — Механизм сбора, 15 — Статус слоя сбора, 16 — Граница интеграции.

Эта статья — рабочий рунбук: конкретные переменные окружения, целевой контракт SQL-вьюхи, референс-класс адаптера, скрипт-форвардер и проверка на /collectors. Где схема источника версионно-зависима (KSC, 1С) — честно помечено, и вынесено в вашу SQL-вьюху / запрос, а не в код.

Конвейер фиксирован и менять его не нужно:

collect() → TelemetryRecord(tabel_no,…)   _tokenize(): anonymize(tabel_no)→user_id   ingest_activity()
   ваш адаптер / форвардер                 (Стрибог, только в enclave-образе)         upsert в activity_facts

TelemetryRecord — ровно 6 полей (контракт Collector, см. ст. 03):

Две точки входа и важное правило гардрейла «ноль внешних обращений»:

Почему так. Тест test_no_outbound_network_imports_in_backend запрещает requests/httpx/ urllib/... в backend/app. SQL-драйвер (pymssql/pyodbc) в этот список НЕ входит → KSC-pull допустим внутри ядра. А любой HTTP-источник (1С OData) ядро тянуть не должен → его выносим в внешний форвардер (PUSH). Это и есть «граница интеграции» из ст. 16, выраженная в коде.

KSC хранит инвентарь и события в БД MS SQL Server (обычно KAV). Стоковая схема не отдаёт напрямую «табельный номер» и «минуты активности по приложению за день» — это собирается из инвентаря KSC + атрибута сотрудника из AD (employeeID→табельный) + событий запуска процессов (Security 4688/4689, если включён аудит). Поэтому маппинг кладём в вашу SQL-вьюху, а адаптер её просто читает.

-- v_lz_app_activity_daily: целевой контракт для коннектора (имена колонок ВАЖНЫ).
-- Источники подставьте под свою версию KSC/AD (ниже — представительный каркас).
CREATE VIEW dbo.v_lz_app_activity_daily AS
SELECT
    ad.employee_id                       AS tabel_no,        -- AD employeeID = табельный
    inv.product_code                     AS sys_code,        -- ваш код системы (или '' → raw_product)
    CONVERT(date, ev.event_time)         AS activity_date,
    COUNT(*)                             AS tx_count,        -- запуски за день (4688)
    SUM(ev.session_seconds) / 60         AS session_minutes, -- из пар 4688/4689
    inv.display_name                     AS raw_product
FROM   dbo.v_lz_process_events ev                            -- ваша вьюха над 4688/4689
JOIN   dbo.v_akpub_host  h   ON h.host_id  = ev.host_id      -- инвентарь хостов KSC
JOIN   dbo.v_lz_host_ad  ad  ON ad.host_id = h.host_id       -- сопоставление хост→сотрудник (AD)
JOIN   dbo.v_lz_inventory inv ON inv.exe   = ev.image_name   -- сопоставление exe→продукт
GROUP BY ad.employee_id, inv.product_code, CONVERT(date, ev.event_time), inv.display_name;

Если у вас уже есть KSC-отчёт/таблица с готовой дневной активностью — оберните её вьюхой с этими 6 колонками. Адаптер ниже не зависит от того, как вы их получили.

backend/app/enclave/ksc_sql_collector.py:

import os
import pymssql                       # SQL-драйвер: НЕ в списке запрещённого egress
from .collectors import TelemetryRecord

class KscSqlCollector:
    source_type = "ksc_sql"

    def collect(self, tenant_id: str, since: str | None = None):
        conn = pymssql.connect(
            server=os.environ["KSC_SQL_HOST"],
            port=int(os.environ.get("KSC_SQL_PORT", "1433")),
            user=os.environ["KSC_SQL_USER"],
            password=os.environ["KSC_SQL_PASSWORD"],
            database=os.environ.get("KSC_SQL_DB", "KAV"),
        )
        sql = """
            SELECT tabel_no, sys_code, activity_date, tx_count, session_minutes, raw_product
            FROM   dbo.v_lz_app_activity_daily
            WHERE  (%s IS NULL OR activity_date >= %s)
        """
        cur = conn.cursor(as_dict=True)
        cur.execute(sql, (since, since))
        for r in cur:
            yield TelemetryRecord(
                tabel_no=str(r["tabel_no"] or ""),
                sys_code=str(r["sys_code"] or ""),
                activity_date=str(r["activity_date"]),
                tx_count=int(r["tx_count"] or 0),
                session_minutes=int(r["session_minutes"] or 0),
                raw_product=str(r["raw_product"] or ""),
            )
        conn.close()

В backend/app/enclave/collectors.py, реестр _REGISTRY:

from .ksc_sql_collector import KscSqlCollector
_REGISTRY = {
    "ksc_sql":     KscSqlCollector(),          # ← было FixtureFileCollector("ksc_sql")
    "odata_1c":    FixtureFileCollector("odata_1c"),
    "syslog_evtx": _syslog_collector(),
}

• В backend/requirements.txt добавить pymssql (или pyodbc + драйвер).
• Переменные окружения коллектора (в .env, не в код/логи/git):

KSC_SQL_HOST, KSC_SQL_PORT, KSC_SQL_DB, KSC_SQL_USER, KSC_SQL_PASSWORD.

• Дать контейнеру collector сетевой доступ к KSC-БД (см. §5).

1С публикует данные по HTTP OData. HTTP-клиент в ядре запрещён гейтом, поэтому 1С подключаем внешним форвардером, который (а) тянет OData у заказчика, (б) обезличивает табельный через Сейф /tokenize, (в) шлёт уже хеши в POST /api/{tenant}/ingest/activity.

GET {ODATA_BASE}/odata/standard.odata/
     InformationRegister_ИсторияРаботыПользователей?$format=json
     &$filter=Period ge datetime'2026-06-01T00:00:00'
Authorization: Basic base64(user:pass)

Имя регистра/реквизиты зависят от конфигурации 1С — подставьте свои; нужны поля «сотрудник/табельный», «дата», «число операций», «длительность».

# forwarder_1c.py — узел интеграции заказчика. Логинимся как collector@<tenant>, берём Bearer.
import os, json, urllib.request

PORTAL = os.environ["LZ_PORTAL_BASE"]          # https://lic.3321616.ru
TENANT = os.environ["LZ_TENANT"]               # напр. artek
TOKEN  = os.environ["LZ_COLLECTOR_TOKEN"]      # JWT роли collector (получить через /api/auth/login)

def _post(path, payload):
    req = urllib.request.Request(
        f"{PORTAL}{path}", data=json.dumps(payload).encode(),
        headers={"Authorization": f"Bearer {TOKEN}", "Content-Type": "application/json"},
    )  # Bearer-запрос НЕ требует CSRF-заголовка (CSRF только для cookie-режима)
    return json.load(urllib.request.urlopen(req))

def tokenize(tabel: str) -> str:
    # обезличиваем в Сейфе → userId-хеш; сырой табельный в портал не уходит
    return _post("/vault/tokenize", {"tenant": TENANT, "tabelNo": tabel})["userId"]

def run(rows_from_1c):
    records = [{
        "userId": tokenize(r["tabel"]),
        "sysCode": r.get("sysCode", ""),
        "activityDate": r["date"],          # YYYY-MM-DD
        "txCount": r["tx"],
        "sessionMinutes": r["minutes"],
        "rawProduct": r.get("product", "1С:Предприятие"),
    } for r in rows_from_1c]
    _post(f"/api/{TENANT}/ingest/activity", {"sourceType": "odata_1c", "records": records})

Получить LZ_COLLECTOR_TOKEN: POST /api/auth/login {"tenant","login":"collector@<tenant>","password"} → поле token. Запускать форвардер по cron у заказчика (например, раз в смену).

Единственный боевой источник, который включается только конфигурацией (класс SyslogFileCollector уже в продукте):

1. Включить аудит создания процессов (GPO: *Audit Process Creation* → события 4688/4689;

при необходимости — командная строка в 4688).

1. Настроить Windows Event Forwarding (WEF) на коллектор-узел или rsyslog, который пишет

события в спул-файл формата RFC 5424 на диск узла коллектора.

1. Задать переменную: SYSLOG_SPOOL_PATH=/var/spool/licenziar/syslog_{tenant}.log

(шаблон {tenant} — раздельные файлы по арендатору).

1. Перезапустить collector. SyslogFileCollector активируется автоматически (иначе — фикстура).

Сетевого egress нет: источник пишет журнал на диск, коллектор читает файл.

Отзыв доступа при увольнении (IdentityConnector.disable_sso) — сетевое действие в AD/SSO, поэтому, как и боевой провижининг, включается явным opt-in и реализуется адаптером под ваш каталог (LDAP/Keycloak). Дефолт — симулятор (ничего наружу). Подробнее — ст. 16 §«Исполнение» и 16 — Жизненный цикл сотрудника.

• Строки подключения/пароли/токены — только в окружении или секрет-менеджере, не в БД, не

в git, не в логах.

• PULL (KSC): контейнеру collector нужен сетевой доступ к KSC-БД; портал/api доступ не нужен.
• PUSH (1С): форвардер стоит у заказчика, наружу ходит только он; ядро остаётся egress-free.
• ANON_SALT держат только enclave-воркеры и Сейф — форвардер обезличивает через Сейф /tokenize,

своей соли не имеет.

• Боевой провижининг (WinRM/SSH/Ansible) и IdentityConnector — двойной opt-in

(PROVISIONER=… + ALLOW_REAL_PROVISIONER=1), иначе старт падает.

1. Поднять/перезапустить контейнер collector с боевыми env.
2. Открыть /collectors — должны появиться прогоны по (арендатор × источник) со статусом

success и applied > 0.

1. Проверить, что цифры пошли: GET /api/{tenant}/ops-efficiency — systemsCovered/среднее

меняются на реальные.

1. Диагностика по таблице симптомов — ст. 04 §«Как читать /collectors»:

0 записей (нет систем у арендатора / нет доступа), skippedUnknownSystem (заполнить raw_product или добавить систему в реестр), прогоны копятся, данные те же (норма — идемпотентность).

Гейт совместимости: любой ваш адаптер обязан пройти тот же контракт-тест формы, что и фикстура (isinstance(adapter, Collector)), — он прогоняется со всем backend-набором. Ядро, приём и статусная машина при подключении не правятся.

Связанные: 03 · 04 · 15 · 16 · 02 — ПДн и Сейф.