Сочный DevOps

Развертываем tetragon с помощью fluxCD

Тетрагон работает по технологии eBPF и позволяет мониторить и даже запрещать разное на уровне ядра Linux. Подробнее тут.

В директории apps создадим директорию tetragon.
Структура этой директории с файлами:

├── helmrelease.yaml
├── kustomization.yaml
├── namespace.yaml
└── repository.yaml

helmrelease.yaml:

apiVersion: helm.toolkit.fluxcd.io/v2beta1
kind: HelmRelease
metadata:
  name: tetragon
  namespace: tetragon
spec:
  interval: 1h0m0s
  releaseName: tetragon
  install: # override existing tetragon CRDs
    crds: CreateReplace
    remediation:
      retries: 3
  upgrade: # update tetragon CRDs
    crds: CreateReplace
  chart:
    spec:
      chart: tetragon
      version: 1.4.0
      sourceRef:
        kind: HelmRepository
        name: cilium
        namespace: flux-system

Без values. Они у каждого свои, но их можно описать прямо в этом файле.

repository.yaml - подключаем репозиторий.

apiVersion: source.toolkit.fluxcd.io/v1
kind: HelmRepository
metadata:
  name: cilium
  namespace: flux-system
spec:
  interval: 1h
  url: https://helm.cilium.io

namespace.yaml - обычный манифест namespace-а.

и kustomization.yaml - там подключаем всё вместе.

apiVersion: kustomize.config.k8s.io/v1beta1
kind: Kustomization
resources:
  - namespace.yaml
  - repository.yaml
  - helmrelease.yaml

Далее у меня в директории clusters описаны разные кластера k8s и в каждой есть директория apps, для "патча" всякого под кластер. В случае tetragon, вот пример patch-values.yaml для тестового кластера:

apiVersion: helm.toolkit.fluxcd.io/v2
kind: HelmRelease
metadata:
  name: tetragon
  namespace: tetragon
spec:
  values:
    tetragon:
      exportAllowList: |-
        {"namespace":["default"],"event_set":["PROCESS_EXEC", "PROCESS_EXIT", "PROCESS_KPROBE", "PROCESS_UPROBE", "PROCESS_TRACEPOINT"]}
      exportDenyList: |-
        {"health_check":true}      
        {"namespace":["", "kube-system", "cloudquery", "flux-system", "kyverno", "logging", "monitoring", "system-serviceaccounts", "keda", "tetragon", "ingress-nginx"]}
      clusterName: "<cluster-name>"
      enableK8sAPI: true
      enableProcessCred: true
      enableProcessNs: false

Мы экспортируем события, связанные с процессами (PROCESS_*), только из namespace default, исключая системные namespace’ы через deny-лист.

И kustomization.yaml:

apiVersion: kustomize.config.k8s.io/v1beta1
kind: Kustomization
resources:
  - ../../../apps/tetragon
  - ../../../policies/tetragon/privileges-raise.yaml
  - ../../../policies/tetragon/dns-only-specified-servers.yaml
  - ../../../policies/tetragon/tcp-listen.yaml
patches:
  - path: patch-values.yaml
    target:
      kind: HelmRelease
      name: tetragon
      namespace: tetragon 

Политики (policies) лежат отдельно на одном уровне с общим apps.
Структура tetragon внутри конкретного кластера.

├── apps
│   ├── tetragon
│   │   ├── kustomization.yaml
│   │   └── patch-values.yaml

#tetragon #fluxcd

Немного про балансировку с haproxy + ingress

Мы используем HAProxy для балансировки трафика на Ingress-ноды Kubernetes. В нашем случае нельзя использовать BGP вместе с MetalLB, поэтому делаем балансировку снаружи — через HAProxy.

Конфигурация HAProxy:

global
    log /dev/log local0
    log /dev/log local1 notice
    daemon
    maxconn 2048

defaults
    log     global
    mode    tcp
    option  tcplog
    timeout connect 10s
    timeout client  1m
    timeout server  1m

frontend k8s_api_frontend
    bind *:8080
    default_backend k8s_api_backend

backend k8s_api_backend
    balance roundrobin
    server master-1 <ip>:6443 check
    server master-2 <ip>:6443 check
    server master-3 <ip>:6443 check

frontend k8s_http_node_port_frontends
    bind *:80
    default_backend k8s_http_node_port_backend

backend k8s_http_node_port_backend
    mode http
    balance roundrobin
    server <ingress-1> <ip>:30080 check
frontend k8s_https_node_port_frontends
    bind *:443
    default_backend k8s_https_node_port_backend

backend k8s_https_node_port_backend
    mode tcp
    balance roundrobin
    server ingress-1 <ip>:30081 check

В Kubernetes на ingress-ноды проброшены NodePort'ы: 30080 для HTTP и 30081 для HTTPS.
Обратите внимание на mode tcp на порту 443. Это необходимо, если вы хотите использовать TLS на уровне Ingress:
1. В mode http HAProxy ожидает видеть обычный HTTP-трафик (открытый текст).
2. Но если клиент подключается по HTTPS, он шлёт зашифрованный трафик, и HAProxy не сможет его прочитать.
3. Поэтому мы используем mode tcp, чтобы не трогать TLS-соединение, а просто прокинуть его до Ingress-контроллера, где оно и будет терминироваться.

Пример ingress:

apiVersion: v1
kind: Ingress
metadata:
  name: <name>
spec:
  ingressClassName: nginx
  rules:
  - host: <domain>
    http:
      paths:
      - backend:
          service:
            name: <svc>
            port:
              number: 80
        path: /
        pathType: Prefix
  tls:
  - hosts:
    - <domain>
    secretName: tls-secret

Таким образом, DNS указывает на HAProxy, который на 443 порту просто прокидывает TLS до Ingress, где и происходит вся логика маршрутизации и терминации.

#haproxy #kubernetes #ingress

Немного про async тесты в python

По долгу службы порой пишешь утилиты на python или небольшие backend системы. К одной из таких решил написать тесты с помощью pytest. Код на fastapi, асинхронный.

Структура:

tests
├── auth
│   └── test_login.py
├── conftest.py

Тут пример с тестом логина, для получения токена. Стандартная JWT авторизация.

В conftest.py:

import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from src.main import app
from asgi_lifespan import LifespanManager


@pytest_asyncio.fixture
async def async_client():
    async with LifespanManager(app):
        transport = ASGITransport(app=app)
        async with AsyncClient(
            transport=transport, base_url="http://testserver"
        ) as client:
            yield client

На реальный сервер запроса не будет, он поднимется внутри pytest, поэтому импортируем app.
Обязательно используйте LifespanManager, если в main.py есть @lifespan или startup/shutdown-логика (например, подключение к БД). Без него — ошибки event loop'а гарантированы.

Далее в тестах:

@pytest.mark.asyncio
async def test_login(async_client):
    response = await async_client.post("/auth/jwt/login", data=...)
    assert response.status_code == 200

#python

Включение kubernetes audit через kubespray на работающем кластере

Добавляем параметры аудита в k8s_cluster/kube_control_plane.yml

# Audit
kubernetes_audit: true
audit_log_path: "/var/log/kubernetes/audit/kube-apiserver-audit.log"
audit_log_maxage: 30
audit_log_maxbackups: 5
audit_log_maxsize: 1000
audit_policy_file: "/etc/kubernetes/audit/policy.yaml"
audit_policy_custom_rules: |

В audit_policy_custom_rules ваша политика.

С прогоном немножко сложно. Дело в том, что если версия k8s не меняется, то kubeadm не перегенерирует статик манифесты подов. Но добавит нужную конфигурацию в kubeadm-config.yml.

Для начал выполним:

ansible-playbook -i inventory --become --become-user=root cluster.yml --limit=kube_control_plane -e "upgrade_cluster_setup=true" --tags master

Это добавит нужное в kubeadm-config.yml.
Далее можно запустить руками на каждом мастер узле генерацию нового статик манифеста для apiserver.

kubeadm init phase control-plane apiserver --config /etc/kubernetes/kubeadm-config.yaml

Или через ansible

ansible -i inventory kube_control_plane -m shell -a "kubeadm init phase control-plane apiserver --config /etc/kubernetes/kubeadm-config.yaml"

После чего на одном из мастеров следует обновить configMap в кластере.

kubeadm init phase upload-config kubeadm --config /etc/kubernetes/kubeadm-config.yaml

Проверяем, что параметры добавились

cat /etc/kubernetes/manifests/kube-apiserver.yaml |grep audit

В выводе будут добавленные параметры.

Проверить логи:

cat /var/log/kubernetes/audit/kube-apiserver-audit.log

#kubernetes

Немного про безопасность подов в k8s

1. Отключение привелигированного режима.

spec:
  containers:
  - name: my-secure-container
    image: myrepo/myapp:latest
    securityContext:
  privileged: false

2. Отключение host namespace.

spec:
   hostPID: false
   hostNetwork: false
   hostIPC: false

3. Отключение монтирование /proc.

 spec:
   containers:
   - name: exploit-container
     image: mymaliciousimage
     procMount: “Default”

4. Не запускать от root пользователя.

 spec:
   containers:
   - name: exploit-container
     image: mymaliciousimage
     securityContext:
       runAsUser: 65534
       runAsNonRoot: true

5. Не использовать wildcard verbs в roles.
6. Корневая файловая система контейнера только для чтения.

 spec:
   containers:
   - name: exploit-container
     image: mymaliciousimage
     securityContext:
       readOnlyRootFilesystem: true

7. Запрещать пуллинг образов из публичных репозиториев.
8. Не использовать в role потенциально опасные verbs

verbs: [‘ímpersonate’, ‘bind’, ‘escalate’]

9. Отключение монтирования sa токена.

spec:
  automountServiceAccountToken: false 

10. Отключение использования latest тега.
11. Отключение эскалации привелегий.

securityContext:
  allowPrivilegeEscalation: false

12. Отключение использования CAP.

securityContext:
        capabilities:
            drop:
                - ALL

13. Отключение использования nginx.ingress.kubernetes.io/server-snippet в ingress.

Все это относительно просто контролировать с помощью kyverno.

#k8s

Получение секретов из vault в ansible

Многие наверное используют lookup плагин community.hashi_vault.hashi_vault для получения секретов из волт.

Например вот так:

kafka_zookeeper_user: "{{ lookup('community.hashi_vault.hashi_vault', 'secret={{ vault_mount }}/{{ vault_secret_path }}/zookeeper/ansible_vars:kafka_zookeeper_user  token={{ vault_token }} url={{ vault_url }}')}}"

И это работает с токеном, у которого расширенные права, но если политика имеет только права на read по нужным путям и ничего более, то с таким токеном будет ошибка permission dined, при условии что у вас kv2 в волт.

Нужно использовать плагин community.hashi_vault.vault_kv2_get. Пример выше превратится в:

kafka_zookeeper_user: "{{ lookup('community.hashi_vault.vault_kv2_get', vault_secret_path ~ '/zookeeper/ansible_vars', engine_mount_point=vault_mount, token=vault_token, url=vault_url).secret['kafka_zookeeper_user'] }}"

Если ранее в переменной vault_mount нужно было использовать data, например secret/data, то теперь не нужно.

#ansible

Динамический сиай в gitlab-ci

У меня структура проекта с ansible выглядит вот так:

Ansible # Группа в gitlab.
   Roles # Группа внутри Ansible, в которой проекты по названию роли.
   Playbooks # Проект с файликами playbook-ов.
   Inventory # Проект с инвентарем. Тут происходит все управление за счет переменных.

Для реализации сиай мы приходим к модели, где любая роль может быть раскатана с теми или иными значениями и поведением за счет управления ее из инвентори, следовательно инвентори это то место, где должна быть реализована сиай.

Есть только один нюанс, с такой точки зрения инвентарь по-сути - это монорепа. Для удобства сделаем сиай динамическим:

default:
  tags: [ <tags> ]

include:
  - project: "<project_path"
    ref: <ci_version>
    file:
      - "<template_file>"

git:get:diff:
  stage: .pre
  extends: [".ansible_git_diff"]
  image: <ansible_image>  
  artifacts:
    reports:
      dotenv: deploy_var.env
    expire_in: 24 hours
    paths:
      - changed_inventories.json
    expire_in: 24 hours

create:deploy:pipeline:
  stage: render
  needs: ["git:get:diff"]
  image: <gomplate_image>
  script:
    - .gitlab/scripts/render.sh
  variables:
    TEMPLATE_FILE: ".gitlab/templates/deploy.yml.tmpl"
    RENDERED_FILE: ".gitlab/dynamic/deploy.yml"
    COLLAPSE_NAME: "Deploy job template"
    COLLAPSE_CAT_FILE: ".gitlab/dynamic/deploy.yml"
  artifacts:
    paths:
      - .gitlab/dynamic
    expire_in: 24 hours

deploy:trigger:
  stage: prepare
  needs: ["git:get:diff", "create:deploy:pipeline"]
  trigger:
    include:
      - artifact: .gitlab/dynamic/deploy.yml
        job: create:deploy:pipeline

Это основной .gitlab-ci.yml в корне инвенторя. Сами темплейты в удаленной репе показаны не будут, они не влезут да и тут только общий подход в конечном проекте.
В джобе create:deploy:pipeline мы рендерим с помощью gomplate вот такой шаблон:

default:
  tags: [ <tags> ]

include:
  - project: "<project_path>"
    ref: <ci_version>
    file:
      - "<template_file>"

{{- if (eq (env.Getenv "SKIP_DEPLOY") "true") }}
nothing:to:deploy:
  stage: deploy
  script:
    - echo "No inventory found for deploy"
{{- else }}
{{- $inventory_dirs := (datasource "inventory_dirs" "changed_inventories.json") }}
{{- range $path := $inventory_dirs }}
deploy:{{ $path | strings.ReplaceAll "/" "-" }}:
  stage: deploy
  extends: [".ansible_run_playbook"]
  image: <ansible_image>
  variables:
    INVENTORY_PATH: {{ $path }}
    PLAYBOOK: {{ index (strings.Split "/" (print $path)) 0 }}.yml
{{- end }}
{{- end }}

Переменную SKIP_DEPLOY мы получали на шаге git:get:diff, как и changed_inventories.json (в нем сравнение изменений по git).
В этой модели предполагается, что директории в инвенторе и имена плэйбуков одинаковы. Это видно по переменным в шаблонной джобе.
Скрипт для рендера выглядит так:

#!/bin/bash

TEMPLATE_DIR="."

function collapse () {
    echo -e "\e[0Ksection_start:`date +%s`:templates[collapsed=true]\r\e[0K$1"
    cat $2
    echo -e "\e[0Ksection_end:`date +%s`:templates\r\e[0K"
}

gomplate -f $TEMPLATE_FILE -o $RENDERED_FILE -d inventory_dirs=$TEMPLATE_DIR
collapse "${COLLAPSE_NAME}" "${COLLAPSE_CAT_FILE}"

Ну и последним шагом deploy:trigger просто вызываем отрендеренный ранее шаблон через trigger job.
В итоге за счет шаблонизации мы получаем динамически создаваемое кол-во джобов в зависимости от изменений в inventory.

#gitlab

Безопасный раннер в gitlab-ci

У нас принято считать безопасным вот такую конфигурацию (на примере нескольких раннеров на одной машине):

concurrent = 10
check_interval = 0

[[runners]]
  name = "docker-ansible"
  url = "<gitlab_url>"
  token = "<project_token>"
  executor = "docker"
  [runners.custom_build_dir]
  [runners.docker]
    image = "<image>"
    privileged = false
    disable_entrypoint_overwrite = false
    oom_kill_disable = false
    disable_cache = false
    shm_size = 0
    volumes = ["/cache"]
    pull_policy = "if-not-present"
  [runners.cache]
    [runners.cache.s3]
    [runners.cache.gcs]
  tag_list = ["<tag>"]
  run_untagged = false
  access_level = "not_protected"
  locked = false
[[runners]]
  name = "docker-ci"
  url = "<gitlab_url>"
  token = "<project_token>"
  executor = "docker"
  [runners.custom_build_dir]
  [runners.docker]
    image = "<image>"
    privileged = false
    disable_entrypoint_overwrite = false
    oom_kill_disable = false
    disable_cache = false
    shm_size = 0
    volumes = ["/cache"]
    pull_policy = "if-not-present"
  [runners.cache]
    [runners.cache.s3]
    [runners.cache.gcs]
  tag_list = ["<tag>"]
  run_untagged = false
  access_level = "not_protected"
  locked = false

Основное это отсутствие привилегированного режима и сервисов. Не монтируем docker.sock и не используем dind (он требует привилегированного режима). Также интернета на раннере нет, только внутренняя сеть. Следовательно мы используем прокси для пакетного менеджера и docker образов.

#gitlab

Версионирование CI шаблонов

У нас репозиторий, где хранится вся CI/CD логика. Как правило так у всех. Репу можно версионировать с помощью bump2version. Для его работы нужен файл VERSION в корне проекта с изначальной версий CI

1.0.0

И конфигурационный файл .bumpversion.cfg:

[bumpversion]
current_version = 1.0.0
commit = True
tag = False

[bumpversion:file:VERSION]

В эти два файла bump2version будет комитить сам, если pipeline запускается из default branch и увеличивать патч версию на один. Во всех остальных случаях, мы просто читаем текущий VERSION, без обновления патч версии и добавляем ему postfix CI_COMMIT_SHORT_SHA, создавая только git tag.

Основной .gitlab-ci.yml:

include:
  - { local: "gitlab/default.yml" }
  - { local: "gitlab/utils.yml"}

stages:
  - bump_and_tag

bump_version_and_tag:
  stage: bump_and_tag
  before_script:
    - !reference [".ci_bump_version", "before_script"]
  script:
    - !reference [".ci_bump_version", "script"]
  rules:
    - if: '$CI_COMMIT_REF_NAME == $CI_DEFAULT_BRANCH'
      when: always
    - if: '$CI_MERGE_REQUEST_IID && $CI_COMMIT_REF_NAME != $CI_DEFAULT_BRANCH'
      when: always
    - if: '$CI_COMMIT_REF_NAME != $CI_DEFAULT_BRANCH && $CI_OPEN_MERGE_REQUESTS'
      when: never
    - if: '$CI_COMMIT_REF_NAME != $CI_DEFAULT_BRANCH'
      when: always
    - when: never              

Подключаемый utils.yml

.ci_bump_version:
  before_script:
    - pip install bump2version
    - git config --global --add safe.directory '*'
    - git config --global user.email "${GITLAB_USER_EMAIL}"
    - git config --global user.name "${GITLAB_USER_NAME}"
    - git remote set-url origin https://tag_token:${TAG_TOKEN}@${CI_SERVER_HOST}/${CI_PROJECT_PATH}.git
    - git fetch
    - git checkout $CI_COMMIT_REF_NAME
    - git reset --hard "origin/$CI_COMMIT_REF_NAME"
  script:
    - |
        if [[ "$CI_COMMIT_REF_NAME" == "$CI_DEFAULT_BRANCH" ]];then
          bump2version patch
          VERSION=$(cat VERSION)
          TAG="v$VERSION"
        else
          VERSION=$(cat VERSION)
          TAG="v$VERSION-$CI_COMMIT_SHORT_SHA"
        fi;
        echo "Generated tag: $TAG"
        git tag -f -a "$TAG" -m "Version $TAG. Created by $GITLAB_USER_NAME"
        git push origin "${TAG}" -o ci.skip
        if [[ "$CI_COMMIT_REF_NAME" == "$CI_DEFAULT_BRANCH" ]];then
          git push origin HEAD:$CI_COMMIT_REF_NAME -o ci.skip
        fi;

Для работы нужно создать ACCESS_TOKEN с правами на api и пуш. Также добавить переменную, например в CI/CD variables с именем TAG_TOKEN.

#gitlab

Использование kubespray как ansible collection

kubespray огромен и тащить его весь к себе в репозиторий смысла нет. Достаточно использовать его как ansible collection и переопределить в inventory только то, что нужно.

Пример структуры inventory:

├── k8s.cluster.1
│   ├── group_vars
│   │   ├── all
│   │   │   ├── all.yml
│   │   │   └── containerd.yml
│   │   ├── k8s_cluster
│   │   │   ├── addons.yml
│   │   │   ├── k8s-cluster.yml
│   │   │   └── kube_control_plane.yml
│   │   └── k8s_load_balancers.yml
│   └── hosts.yml
├── k8s.cluster.2
│   ├── group_vars
│   │   ├── all
│   │   │   ├── all.yml
│   │   │   └── containerd.yml
│   │   ├── k8s_cluster
│   │   │   ├── addons.yml
│   │   │   ├── k8s-cluster.yml
│   │   │   └── kube_control_plane.yml
│   │   └── k8s_load_balancers.yml
│   └── hosts.yml
├── README.md
├── requirements.txt
└── requirements.yml

По похожей схеме описываем любые кластера. Имена файлов такие же как и у kubespray, не считая k8s_load_balancers, это часть с реального проекта, но в данном посте можно не обращать на нее внимания.

В requirements.yml:

collections:
- name: https://github.com/kubernetes-sigs/kubespray
  type: git
  version: v2.27.0

Устанавливаем коллекцию:

ansible-galaxy install -r inventory/kubernetes/requirements.yml

В requirements.txt оригинальные python зависимости из kubespray:

ansible==9.13.0
# Needed for community.crypto module
cryptography==44.0.2
# Needed for jinja2 json_query templating
jmespath==1.0.1
# Needed for ansible.utils.ipaddr
netaddr==1.3.0

Далее дефолтный флоу из доки, для сетапа окружения:

python3 -m venv .kubespray
source .kubespray/bin/activate
pip install -U -r inventory/kubernetes/requirements.txt
pip install ruamel.yaml

Если у вас установлен ansible через пакетный менеджер дистрибутива, то могут возникнуть проблемы с путями до модулей. Пофиксить можно так:

export ANSIBLE_LIBRARY=./.kubespray/lib/python3.11/site-packages/ansible/modules
export ANSIBLE_MODULE_UTILS=./.kubespray/lib/python3.11/site-packages/ansible/module_utils
export PYTHONPATH=./.kubespray/lib/python3.11/site-packages:$PYTHONPATH

Версия питона само собой может отличаться.

Заполняем hosts.yml. Вообще в доке старый ini формат. Но я предпочитаю yml. Пример:

all:
  hosts:
    master-1:
      ansible_host: <dns_name>
      ip: <ip_address>
      access_ip: <access_ip>
    worker-1:
      ansible_host: <dns_name>
      ip: <ip_address>
      access_ip: <access_ip>
      node_labels:
        node-role.kubernetes.io/worker: ""
    ingress-1:
      ansible_host: <dns_name>
      ip: <ip_address>
      access_ip: <access_ip>
      node_labels:
        node-role.kubernetes.io/ingress: ""
      node_taints:
        - "node-role.kubernetes.io/ingress=:NoSchedule"
  children:
    kube_control_plane:
      hosts:
        master-1:
    kube_node:
      hosts:
        worker-1:
        ingress-1:
    etcd:
      hosts:
        master-1:
    k8s_cluster:
      children:
        kube_control_plane:
        kube_node:
    calico_rr:
      hosts: {}

Установить кластер:

ansible-playbook -i inventory/kubernetes/<cluster>/hosts.yml --become --become-user=root playbooks/kubernetes.yml

Мой playbook:

- name: Install Kubernetes
  ansible.builtin.import_playbook: kubernetes_sigs.kubespray.cluster

Чтобы добавить новую ноду, сначала добавить вот такой playbook:

- name: Gathering facts
  ansible.builtin.import_playbook: kubernetes_sigs.kubespray.facts
  tags: facts
  when: "'facts' in ansible_run_tags"
  
- name: Scale the kubernetes cluster
  ansible.builtin.import_playbook: kubernetes_sigs.kubespray.scale

Перед установкой рекомендуют собрать факты:

ansible-playbook -i inventory/kubernetes/<cluster>/hosts.yml --become --become-user=root playbooks/kuberentes_scale.yml --tags="facts"

И затем добавить ноду:

ansible-playbook -i inventory/kubernetes/<cluster>/hosts.yml --become --become-user=root playbooks/kuberentes_scale.yml --limit=<node_name>

Перед этим не забыв добавить ее по аналогии в hosts.yml

#kubernetes

Интеграция kafbat и keycloak

Сначала на стороне keycloak нужно создать клиента. И обязательно создать маппер, который собственно за маппит ваши роли на верхней уровень, например roles. JSON должен быть примерно таким:

"roles": [
  "<role1>",
  "<role2>",
],

Допустим у нас будет две роли, admins, viewers. В kafka-ui.yaml нужно добавить следующее:

auth:
  type: OAUTH2
  oauth2:
    client:
      keycloak:
        clientId: <clientId>
        clientSecret: <clientSecret>
        scope: openid
        client-name: keycloak
        provider: keycloak
        redirect-url: https://<kafka_ui_domain>/login/oauth2/code/keycloak
        authorization-grant-type: authorization_code
        issuer-uri: https://<keycloak_domain>/realms/<realm_name>
        user-name-attribute: preferred_username
        custom-params:
          type: oauth
          roles-field: roles

Это касательно интеграции с keycloak. Поле roles-field как раз отвечает за получение из токена ролей и да, оно не умеет брать их из вложенной структуры вида resource_access.<client_id>.roles. Поэтому мы создали маппер выше.

Далее подключаем там же маппинг ролей уже на нашей стороне:

rbac:
  roles:
    - name: "admins"
      clusters:
        - <cluster_name_1>
        - <cluster_name_2>
      subjects:
        - provider: oauth
          type: role
          value: "admins" ## Имя роли в KK

      permissions:
        - resource: applicationconfig
          actions: all

        - resource: clusterconfig
          actions: all

        - resource: topic
          value: ".*"
          actions: all

        - resource: consumer
          value: ".*"
          actions: all

        - resource: schema
          value: ".*"
          actions: all

        - resource: connect
          value: ".*"
          actions: all

        - resource: ksql
          actions: all

        - resource: acl
          actions: all

        - resource: audit
          actions: all

    - name: "viewers"
      clusters:
        - <cluster_name_1>
        - <cluster_name_2>
      subjects:
        - provider: oauth
          type: role
          value: "viewers" ## Имя роли в KK
      permissions:
        - resource: clusterconfig
          actions: [ "view" ]

        - resource: topic
          value: ".*"
          actions:
            - VIEW
            - MESSAGES_READ

        - resource: consumer
          value: ".*"
          actions: [ view ]

        - resource: schema
          value: ".*"
          actions: [ view ]

        - resource: connect
          value: ".*"
          actions: [ view ]

        - resource: acl
          actions: [ view ]

Любой залогиненый в KK пользователь получит пустой дашборд без возможности что-либо сделать. Пользователи добавленные в соответствующие группы в КК получат тот доступ, который мы указали в kafka-ui выше.

#kafka

Пример реализации GitOps на fluxCD

fluxCD позволяет реализовать GitOps на kustomize + sops, если вдруг нет vault webhook или что-то вроде.

Первым делом сделаем bootstrap в кластере. Нужно быть залогиненым в кластер или передать kubeconfig через --kubeconfig.

flux bootstrap git \
  --url=ssh://git@<domain>/path/flux.git \
  --branch=main \
  --path=clusters/k8s-1 \
  --private-key-file=<path/to/private_key>

Это установит нужные приложения в k8s.
Тоже самое можно сделать для кластера k8s-2 например. Итоговая структура с учетом нескольких приложений.

├── clusters
│   ├── apps
│   │   ├── keda
│   │   │   ├── helmrelease.yaml
│   │   │   ├── kustomization.yaml
│   │   │   ├── namespace.yaml
│   │   │   └── repository.yaml
│   │   └── teleport-agent
│   │       ├── clusterrolebinding.yaml
│   │       ├── helmrelease.yaml
│   │       ├── kustomization.yaml
│   │       ├── namespace.yaml
│   │       ├── repository.yaml
│   │       └── token.yaml
│   ├── k8s-1
│   │   ├── apps
│   │   │   └── kustomization.yaml
│   │   └── flux-system
│   │       ├── gotk-components.yaml
│   │       ├── gotk-sync.yaml
│   │       └── kustomization.yaml
│   └── k8s-2
│       ├── apps
│       │   ├── kustomization.yaml
│       │   └── teleport-agent
│       │       ├── kustomization.yaml
│       │       └── patch-values.yaml
│       └── flux-system
│           ├── gotk-components.yaml
│           ├── gotk-sync.yaml
│           └── kustomization.yaml
└── README.md

Собственно на верхний уровень вынесены общие для кластеров приложения. В этом примере keda и teleport-agent.

В телепорте нужно под кластер переопределять values + использовать секрет, который не нужно хранить в репе.
Собственно для секрета один из способов, это использование sops.
Я использую в качестве бэка sops - age.
Сформировать ключ можно так:

age-keygen -o ~/age.key

Далее завести в кластер:

kubectl create secret generic sops-age --namespace=flux-system --from-file=age.agekey=/path/to/age_private_key

В нашей структуре создаем secret вида:

apiVersion: v1
kind: Secret
metadata:
  name: teleport-agent-token
  namespace: teleport-agent
stringData:
  values.yaml: |
    authToken: "super-secret-token"

Странный ключ values.yaml нужен для flux, иначе будет ошибка. А вот ключ authToken уже пойдет в переменную нужную чарту teleport.

Можно добавить .sops.yaml в корень репозитория:

creation_rules:
  - path_regex: clusters/apps/.*/.*\.yaml
    encrypted_regex: '^(data|stringData)$'
    age: '<age_public_key>'

И шифруем его sops-ом:

sops -e -i <path/to/secret>

В итоге в репозитории окажется зашифрованный файл. Flux из коробки умеет работать с sops, нужно только в gotk-sync.yaml указать ему это например так:

 apiVersion: kustomize.toolkit.fluxcd.io/v1
kind: Kustomization
metadata:
  name: flux-system
  namespace: flux-system
spec:
  interval: 10m0s
  path: ./clusters/k8s-1
  prune: true
  sourceRef:
    kind: GitRepository
    name: flux-system
  decryption:
    provider: sops
    secretRef:
      name: sops-age

sops-age - секрет, который мы создали.

Касательно оверайда values. В кластерах нам нужно мочь переопределить значения в values под кластер для телепорт агента.
Создаем директорию teleport-agent внутри clusters/<cluster>/apps.

Там нужно два файла, собственно kustomization.yaml

apiVersion: kustomize.config.k8s.io/v1beta1
kind: Kustomization
resources:
  - ../../../apps/teleport-agent
patchesStrategicMerge:
  - patch-values.yaml

В нем подключаем общую директорию и указываем смержить с файлом path-values.yaml.
Собственно patch-values.yaml:

apiVersion: helm.toolkit.fluxcd.io/v2
kind: HelmRelease
metadata:
  name: teleport-agent
  namespace: teleport-agent
spec:
  values:
    kubeClusterName: k8s-1
    proxyAddr: <teleport_server_address>
    labels:
      team: <team>
      env: <env>
      name: <cluster_name>

Добавили spec.values. В общем файле его нет. Там мы подключаем переменную из секрета через valuesFrom.

При комите в целевую ветку, flux сам создаст релиз и подхватит деплой с kustomization + sops.

#flux

Подключение vault datasource в gomplate шаблонах

Предположим что в шаблонизированном файле нам нужно получать переменные из vault и подставлять их собственно в шаблон.

Пример шаблона:

{{- $vaultPath := "secret/test/env" }}
var: {{ (ds "vault" $vaultPath).KEY_NAME_IN_VAULT }}

Отрендерить можно командой:

gomplate -d "vault=vault+$VAULT_ADDR?token=$VAULT_TOKEN" \
      -f <input_file> \
      -o <output_file

Также экспортировать переменные окружения VAULT_ADDR и VAULT_TOKEN.

#gomplate

DevSecOps roadmap

Интересный репозиторий с роадмапом для будущих DevSecOps инженеров.
Всякие гайды по DevSecOps от самого OWASP.
Еще есть бесплатный курс от gitlab.

#обучение

Переезд на новые ZooKeeper-серверы

Появилась задачка мигрировать ZooKeeper-сервера на новые, при этом Kafka должна продолжать работать.
Просто взять и подменить адреса в server.properties — не вариант: Kafka хранит в ZooKeeper своё состояние, и при переключении на “чистый” кластер всё развалится.
Перенос dataDir с остановкой тоже не подходит — будет простой.

Есть третий, рабочий способ: расширяем текущий кластер ZooKeeper новыми нодами и потом исключаем старые. Рассказываю, как сделал это у себя.

Вводные:
Был ZooKeeper-кластер из 3 старых нод.
Новые — тоже 3, плюс одна временная для обеспечения кворума.
Новые сервера за NAT: внешние адреса доступны со старых, но не существуют локально на новых машинах.

Подготовка:
На старых нодах в zoo.cfg включаем:

reconfigEnabled=true

А в zkEnv.sh добавляем временно:

export SERVER_JVMFLAGS="$SERVER_JVMFLAGS -Dzookeeper.skipACL=yes"

Это нужно, чтобы не возиться с ACL при реконфигурации.

Добавляем новые ноды:
Разворачиваем ZooKeeper на новых машинах (как угодно — вручную, Ansible), но не запускаем его.

На старой ZooKeeper-ноде:

zkCli.sh
reconfig -add server.4=<dns_name>:2888:3888:participant;2181

Указываем DNS-имя новой ноды (резолвится во внешний IP).

ZooKeeper создаёт новый zookeeper.cfg.dynamic.<id> и прописывает его в zoo.cfg автоматически.
Запускаем ZooKeeper на новой, только что добавленной ноде.

Проблема с bind’ом:
Когда запускаем zk-4, видим, что он не может создать сокет на внешнем IP, ведь он не существует на машине (из-за NAT).
Поэтому на новой ноде руками правим zookeeper.cfg.dynamic.<id>, заменяя внешний адрес на внутренний, локальный IP ноды (из NAT-сети).

Повторяем для остальных нод
Каждый раз при reconfig -add создаётся новый dynamic config, и снова приходится вручную править IP-адреса на новых серверах.

Переключение Kafka
Когда в ZooKeeper уже 7 нод (3 старых + 4 новых), на Kafka-брокерах в server.properties указываем новые адреса ZooKeeper, а старые пока оставляем:

zookeeper.connect=new1:2181,new2:2181,new3:2181,old1:2181,old2:2181

Рестартим Kafka.

Удаление старых нод
На одной из новых нод:

zkCli.sh
reconfig -remove <old_id>

Удаляем старые ноды по myid.

ZooKeeper снова создаст новый конфиг — и его снова нужно поправить на новых машинах (внутренние IP вместо внешних).

Удаляем старые ZooKeeper-адреса из server.properties Kafka
Останавливаем старые ZooKeeper-ноды

На новых ZooKeeper:
Убираем skipACL
Можно убрать reconfigEnabled, если не нужен
В итоге переезд выполнен без downtime на работающем кластере kafka.

#zookeeper

Кастомные функции в ansible

В Ansible, помимо модулей, можно писать свои фильтры — filter_plugins. Это по сути обычные Python-функции, которые можно вызывать в шаблонах и тасках.

Например, у нас есть Kafka-роль, в которой понадобилось добавить возможность менять фактор репликации (replication factor). Операция редкая, но почему бы сразу не сделать через IaC?

Kafka предоставляет утилиту kafka-reassign-partitions.sh, которая формирует JSON с текущим и предложенным состоянием.

Примерная команды:

kafka-reassign-partitions.sh  --bootstrap-server <ip>:<port> --command-config admin_connect.cfg --generate --topics-to-move-json-file /tmp/kafka_topics_to_move.json --broker-list "1,2,3"

broker-list это id брокеров, не их IP адреса.

В итоге мы получим что-то вроде:

Current partition replica assignment
{"version":1,"partitions":[{"topic":"test-test-test","partition":0,"replicas":[3],"log_dirs":["any"]},{"topic":"test-test-test","partition":1,"replicas":[1],"log_dirs":["any"]},{"topic":"test-test-test","partition":2,"replicas":[2],"log_dirs":["any"]}]}

Proposed partition reassignment configuration
{"version":1,"partitions":[{"topic":"test-test-test","partition":0,"replicas":[2],"log_dirs":["any"]},{"topic":"test-test-test","partition":1,"replicas":[3],"log_dirs":["any"]},{"topic":"test-test-test","partition":2,"replicas":[1],"log_dirs":["any"]}]}

Отсюда нам полезно взять то, что после Proposed. В ansible это можно сделать так:

- name: Extract JSON from reassignment plan
  set_fact:
    proposed_json: "{{ reassignment_plan.stdout.split('Proposed partition reassignment configuration')[-1] | trim | from_json }}"

А дальше уже интереснее — логичнее использовать кастомную фильтр-функцию, чтобы перераспределить реплики с нужным RF.

Создадим в корне роли директорию filter_plugins с файлом set_new_rf.py:

from typing import Dict, Any
import itertools

class FilterModule(object):
    """Custom filter to set replication factor in kafka."""

    def filters(self):
        return {
            'set_rf': self.set_rf,
        }

    def set_rf(self, proposed_json: Dict[str, Any], new_rf: int, kafka_brokers: str) -> Dict[str, Any]:
        """Return copy of proposed_json with new replication factor"""
        updated_partitions = []
        brokers = [int(x) for x in kafka_brokers.split(',')]
        broker_cycle = itertools.cycle(brokers)

        for partition in proposed_json.get('partitions', []):
            replicas = partition['replicas']
            unique_replicas = list(set(replicas))
            length = len(unique_replicas)

            if new_rf <= length:
                new_replicas = sorted(unique_replicas, key=lambda x: brokers.index(x))[:new_rf]
            else:
                needed = new_rf - length
                possible_kafka_brokers = [broker for broker in brokers if broker not in unique_replicas]
                new_replicas = unique_replicas + possible_kafka_brokers[:needed]
            
            if new_rf == 1:
                new_replicas = [next(broker_cycle)]

            updated_partitions.append({
                'topic': partition['topic'],
                'partition': partition['partition'],
                'replicas': new_replicas
            })

        return {
            'version': proposed_json.get('version', 1),
            'partitions': updated_partitions
        }

Код сформирует json с равномерным распределением партиций как при увеличении RF, так и при снижении.

Вызвать в таске можно например так:

- name: Modify proposed json via custom filter
  set_fact:
    updated_proposed_json: "{{ proposed_json | set_rf(NEW_RF|int, kafka_broker_list) }}"

Остается только выполнить execute:

kafka-reassign-partitions.sh --bootstrap-server <ip>:<port> --execute --reassignment-json-file /tmp/kafka_updated_proposed.json --command-config admin_connect.cfg

#ansible

Настроечка kafka-ui для подключение нескольких кластеров

Для начала нам нужен docker-compose файл, примерно следующего содержания:

services:
  kafka-ui:
    container_name: kafka-ui
    image: provectuslabs/kafka-ui:latest
    environment:
      DYNAMIC_CONFIG_ENABLED: true
    volumes:
    - /etc/kafka-ui/kafka-ui.yaml:/etc/kafkaui/dynamic_config.yaml
    - <kafka_cluster_1>truststore.jks:/etc/kafkaui/<kafka_cluster_1>truststore.jks
    - <kafka_cluster_2>truststore.jks:/etc/kafkaui/<kafka_cluster_2>truststore.jks
    - <kafka_cluster_3>truststore.jks:/etc/kafkaui/<kafka_cluster_3>truststore.jks

Если используем SSL, то обязательно монтируем java truststore.jks нужных кластеров.

Затем файлик kafka-ui.yaml:

auth:
  type: LOGIN_FORM

spring:
  security:
    user:
      name: <ui_user>
      password: <ui_password>

kafka:
  clusters:
    - name: kafka-cluster-1
      bootstrapServers: <bootstrap_hosts>:<ssl_port>
      properties:
        security.protocol: SASL_SSL
        sasl.mechanism: SCRAM-SHA-256
        sasl.jaas.config: org.apache.kafka.common.security.scram.ScramLoginModule required username="<user>" password="<password>";
        ssl.truststore.location: /etc/kafkaui/<kafka_cluster_1>truststore.jks
        ssl.truststore.password: <truststore_pass>
        ssl.truststore.type: JKS

    - name: kafka-cluster-2
      bootstrapServers: <bootstrap_hosts>:<ssl_port>
      properties:
        security.protocol: SASL_SSL
        sasl.mechanism: SCRAM-SHA-256
        sasl.jaas.config: org.apache.kafka.common.security.scram.ScramLoginModule required username="<user>" password="<password>";
        ssl.truststore.location: /etc/kafkaui/<kafka_cluster_2>truststore.jks
        ssl.truststore.password: <truststore_pass>
        ssl.truststore.type: JKS

    - name: kafka-cluster-3
      bootstrapServers: <bootstrap_hosts>:<ssl_port>
      properties:
        security.protocol: SASL_SSL
        sasl.mechanism: PLAIN
        sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="<user>" password="<password>";
        ssl.truststore.location: /etc/kafkaui/<kafka_cluster_3>truststore.jks
        ssl.truststore.password: <truststore_pass>
        ssl.truststore.type: JKS

rbac:
  roles: []

webclient: {}

В примере мы подключаем 3 кластера, все используют SSL, два из них на SCRAM аутентификации, обычно с ней хранят пользователей в zookeeper. Один кластер использует PLAIN.

По примеру можно подключать и другие кластера, меняя те или иные настройки, например SSL или механизмы аутентификации.

spring.security.user - это креды от входа в kafka-ui.
bootstrap сервера можно перечислять через запятую, но достаточно и одного.

#kafka

Добавление нового listener в kafka

Допустим есть кластер кафки, который использует в качестве механизма аутентификации SCRAM-SHA-256. Но тут понадобилось добавить одного listener под SASL_PLAINTEXT. Сделать это можно довольно просто.

Добавляем нового listener в listeners:

listeners=NEW://<ip>:<new_port>

Затем в advertised.listeners:

advertised.listeners=NEW://<ip>:<new_port>

И указываем маппинг:

listener.security.protocol.map=NEW:SASL_PLAINTEXT

Клиенту говорим следующие параметры:
1. security protocol = SASL_PLAINTEXT
2. sasl.mechanism = SCRAM-SHA-256

И креды от пользователя и название consumer group.
Теперь у вас Kafka доступна по двум listener'ам — с разными протоколами, но с одной и той же схемой аутентификации.
#kafka

API honeypot на FastApi

Ссылка на проект тут.
Хонипот основан на библиотеке baitroute.
Суть работы очень простая, из директории rules подгружаются разные как-бы уязвимые endpoints, которые могут обрабатывать GET, POST, HEAD запросы, в зависимости от конфигурации.

Запустить и потыкать можно так:

poetry install --no-root
python -m src.main

Парсим стандартный nginx access log vector-ом и раскладываем по ECS

ECS (Elastic Common Schema) - это стандарт для унификации структурированных данных. Он определяет, как должны выглядеть поля логов, что упрощает их обработку, анализ и визуализацию. Более подробно тут.

Конфиг парсинга:

observer.product = "<Product name>"
.observer.vendor = "<Vendor name>"
.event.kind = "event"
.event.category = ["web"]
.event.type = ["access", "info
structured, json_err = parse_json(.message)
if json_err == null && is_object(structured) {
  . = merge!(., structured)
} else {
  .parsed, re_err = parse_regex(
    .message,
    pattern: r'(?P<source_ip>[\d\.]+) - - \[(?P<time>.+?)\] "(?P<method>[A-Z]+) (?P<path>[^ ]+) HTTP/(?P<http_version>[0-9\.]+)" (?P<status_code>\d+) (?P<bytes_sent>\d+|-) "(?P<referrer>[^"]*)" "(?P<user_agent>[^"]*)" "(?P<extra>[^"]*)"'
  )
  if re_err == null && is_object(.parsed) {
    .source.ip = .parsed.source_ip
    .http.request.method = .parsed.method
    .url.path = .parsed.path
    .http.version = .parsed.http_versi
    if exists(.parsed.status_code) {
      code, parse_err = parse_int(.parsed.status_code)
      if parse_err != null {
        code = -1
      }
      .http.response.status_code = code
    } else {
      .http.response.status_code = -1
  
    if .http.response.status_code >= 200 && .http.response.status_code <= 299 {
      .event.outcome = "success"
    } else if .http.response.status_code >= 400 && .http.response.status_code <= 499 {
      .event.outcome = "failure"
    } else if .http.response.status_code >= 500 && .http.response.status_code <= 599 {
      .event.outcome = "failure"
    } else {
      .event.outcome = "unknown"
  
    .user_agent.original = .parsed.user_agent
    .event.created, ts_err = parse_timestamp(
      to_string(.parsed.time),
      format: "%d/%b/%Y:%H:%M:%S %z"
    )
    if ts_err != null {
      .event.created = null
    }
    del(.parsed.extra)
    del(.parsed)
  }
}
del(.message)

Сначала пробуем распапрсить .message как JSON. Если парсинг JSON не удался, то применяем regexp для разбиения строки на поля. Далее полученные данные приводим к формату ECS + небольшая логика, чтобы выставить правильно .event.outcome поле.

#vector