PGSync错误,连接到elasticsearch时发生身份验证异常(401):“无法为REST请求[/]对用户[elastic]进行身份验证”

PGSync版本: 3.1.0

Postgres 版本:16.1

Elasticsearch版本: 弹性搜索8.11.1

Redis版本: 7.2.3

Python版本: 3.11.7

问题描述: 我在弹性搜索身份验证方面遇到了问题,尽管我设置了相同的密码和SSL证书,这三个证书:客户端证书、CA证书和密钥证书。如果我关闭ssl和https作为模式,一切都很好,但我需要使用安全连接。下面是一个可复制的例子。我可以通过kibana和curl登录到elasticsearch,这很好(我还没有测试curl提供ca和客户端证书)。

如有任何反馈,我们将不胜感激。

Docker YAML文件:

version: '3.9'

services:
  setup:
    networks:
      - elastic
    image: elasticsearch:${STACK_VERSION}
    volumes:
      - certs:/usr/share/elasticsearch/config/certs
    user: '0'
    command: >
      bash -c '
        if [ x${ELASTIC_PASSWORD} == x ]; then
          echo "Set the ELASTIC_PASSWORD environment variable in the .env file";
          exit 1;
        elif [ x${KIBANA_PASSWORD} == x ]; then
          echo "Set the KIBANA_PASSWORD environment variable in the .env file";
          exit 1;
        fi;
        if [ ! -f config/certs/ca.zip ]; then
          echo "Creating CA";
          bin/elasticsearch-certutil ca --silent --pem -out config/certs/ca.zip;
          unzip config/certs/ca.zip -d config/certs;
        fi;
        if [ ! -f config/certs/certs.zip ]; then
          echo "Creating certs";
          echo -ne \
          "instances:\n"\
          "  - name: es01\n"\
          "    dns:\n"\
          "      - es01\n"\
          "      - localhost\n"\
          "    ip:\n"\
          "      - 127.0.0.1\n"\
          > config/certs/instances.yml;
          bin/elasticsearch-certutil cert --silent --pem -out config/certs/certs.zip --in config/certs/instances.yml --ca-cert config/certs/ca/ca.crt --ca-key config/certs/ca/ca.key;
          unzip config/certs/certs.zip -d config/certs;
        fi;
        echo "Setting file permissions"
        chown -R root:root config/certs;
        find . -type d -exec chmod 750 \{\} \;;
        find . -type f -exec chmod 640 \{\} \;;
        echo "Waiting for Elasticsearch availability";
        until curl -s --cacert config/certs/ca/ca.crt https://es01:9200 | grep -q "missing authentication credentials"; do sleep 30; done;
        echo "Setting kibana_system password";
        until curl -s -X POST --cacert config/certs/ca/ca.crt -u "elastic:${ELASTIC_PASSWORD}" -H "Content-Type: application/json" https://es01:9200/_security/user/kibana_system/_password -d "{\"password\":\"${KIBANA_PASSWORD}\"}" | grep -q "^{}"; do sleep 10; done;
        echo "All done!";
      '
    healthcheck:
      test: ['CMD-SHELL', '[ -f config/certs/es01/es01.crt ]']
      interval: 1s
      timeout: 5s
      retries: 120

  es01:
    depends_on:
      setup:
        condition: service_healthy
        
    networks:
      - elastic
      - frontend
      - data-center
      
    image: elasticsearch:${STACK_VERSION}
    volumes:
      - certs:/usr/share/elasticsearch/config/certs
      - esdata01:/usr/share/elasticsearch/data
    ports:
      - ${ES_PORT}:9200
    environment:
      - node.name=es01
      - cluster.name=${CLUSTER_NAME}
      - cluster.initial_master_nodes=es01
      - ELASTIC_PASSWORD=${ELASTIC_PASSWORD}
      - bootstrap.memory_lock=true
      - xpack.security.enabled=true
      - xpack.security.http.ssl.enabled=true
      - xpack.security.http.ssl.key=certs/es01/es01.key
      - xpack.security.http.ssl.certificate=certs/es01/es01.crt
      - xpack.security.http.ssl.certificate_authorities=certs/ca/ca.crt
      - xpack.security.transport.ssl.enabled=true
      - xpack.security.transport.ssl.key=certs/es01/es01.key
      - xpack.security.transport.ssl.certificate=certs/es01/es01.crt
      - xpack.security.transport.ssl.certificate_authorities=certs/ca/ca.crt
      - xpack.security.transport.ssl.verification_mode=certificate
      - xpack.license.self_generated.type=${LICENSE}
    mem_limit: ${MEM_LIMIT}
    ulimits:
      memlock:
        soft: -1
        hard: -1
    healthcheck:
      test:
        [
          'CMD-SHELL',
          "curl -s --cacert config/certs/ca/ca.crt https://localhost:9200 | grep -q 'missing authentication credentials'",
        ]
      interval: 10s
      timeout: 10s
      retries: 120

  kibana:
    depends_on:
      es01:
        condition: service_healthy
        
    networks:
      - elastic
      
    image: kibana:${STACK_VERSION}
    
    volumes:
      - certs:/usr/share/kibana/config/certs
      - kibanadata:/usr/share/kibana/data
      
    ports:
      - ${KIBANA_PORT}:5601
      
    environment:
      - SERVERNAME=kibana
      - ELASTICSEARCH_HOSTS=https://es01:9200
      - ELASTICSEARCH_USERNAME=kibana_system
      - ELASTICSEARCH_PASSWORD=${KIBANA_PASSWORD}
      - ELASTICSEARCH_SSL_CERTIFICATEAUTHORITIES=config/certs/ca/ca.crt
      
    mem_limit: ${MEM_LIMIT}
    healthcheck:
      test:
        [
          'CMD-SHELL',
          "curl -s -I http://localhost:5601 | grep -q 'HTTP/1.1 302 Found'",
        ]
      interval: 10s
      timeout: 10s
      retries: 120

  postgres:
    ports:
      - 5432:5432
    image: postgres:16.1
    volumes:
      - "./init_db:/docker-entrypoint-initdb.d"
      # - "../postgres/data:/var/lib/postgresql/data"
    environment:
      - "POSTGRES_USER=revista"
      - "POSTGRES_PASSWORD=${ELASTIC_PASSWORD}"
    restart: always

    networks:
      - data-center
    
    command: 
      - "postgres"
      - "-c"
      - "wal_level=logical"
      - "-c"
      - "max_replication_slots=3"
    
  redis:
    image: redis:7.2.3
    ports:
      - 6379:6379
    command: redis-server --requirepass ${ELASTIC_PASSWORD}
    networks:
      - data-center
    
  pgsync:
    build:
      context: ../pgsync
      dockerfile: ./Dockerfile-pgsync

    restart: on-failure
    
    labels:
      org.label-schema.name: "pgsync"
      org.label-schema.description: "Postgres to Elasticsearch sync"
      com.label-schema.service-type: "daemon"
      
    sysctls:
      - net.ipv4.tcp_keepalive_time=200
      - net.ipv4.tcp_keepalive_intvl=200
      - net.ipv4.tcp_keepalive_probes=5
      
    depends_on:
      - postgres
      - redis
      
    environment:
      - CHECKPOINT_PATH=/pgsync
      - PG_USER=revista
      - PG_HOST=postgres
      - PG_PORT=5432
      - PG_PASSWORD=${ELASTIC_PASSWORD}
      - ELASTICSEARCH_SCHEME=https
      - ELASTICSEARCH_HOST=es01
      - ELASTICSEARCH_PORT=9200
      - ELASTICSEARCH_USER=elastic
      - ELASTICSEARCH_PASSWORD=${ELASTIC_PASSWORD}
      - ELASTICSEARCH_VERIFY_CERTS=True
      - ELASTICSEARCH_USE_SSL=True
      - REDIS_HOST=redis
      - REDIS_PORT=6379
      - REDIS_AUTH=${ELASTIC_PASSWORD}
      - ELASTICSEARCH=True
      - OPENSEARCH=False
      - SCHEMA=/pgsync/schema.json
      - LOG_LEVEL=INFO
      - ELASTICSEARCH_CA_CERTS=/usr/share/elasticsearch/config/certs/ca/ca.crt
      - ELASTICSEARCH_CLIENT_CERT=/usr/share/elasticsearch/config/certs/es01/es01.crt
      - ELASTICSEARCH_CLIENT_KEY=/usr/share/elasticsearch/config/certs/es01/es01.key


    command: ./runserver.sh && sleep inf

    volumes:
      # - ../pgsync/example-schema.json:/pgsync/schema.json
      - ../pgsync/schema.json:/pgsync/schema.json
      - certs:/usr/share/elasticsearch/config/certs

    networks:
      - data-center

volumes:
  certs:
    driver: local
  esdata01:
    driver: local
  kibanadata:
    driver: local

networks:
  networks:
  data-center:
  frontend:
  elastic:

以下是我用于构建pgsync映像的Dockerfile:

FROM python:3.11
ARG WORKDIR=/pgsync
RUN mkdir $WORKDIR
WORKDIR $WORKDIR
RUN pip install pgsync
COPY scripts/wait-for-it.sh wait-for-it.sh
COPY scripts/runserver.sh runserver.sh
RUN chmod +x wait-for-it.sh
RUN chmod +x runserver.sh
CMD ./runserver.sh && sleep inf

这两个脚本都出现在Dockerfile中,并推荐在项目github的页面上(链接: https://github.com/toluaina/pgsync ):

等待-it.sh

#!/usr/bin/env bash
# Use this script to test if a given TCP host/port are available

cmdname=$(basename $0)

echoerr() { if [[ $QUIET -ne 1 ]]; then echo "$@" 1>&2; fi }

usage()

{
    cat << USAGE >&2
Usage:
    $cmdname host:port [-s] [-t timeout] [-- command args]
    -h HOST | --host=HOST       Host or IP under test
    -p PORT | --port=PORT       TCP port under test
                                Alternatively, you specify the host and port as host:port
    -s | --strict               Only execute subcommand if the test succeeds
    -q | --quiet                Don't output any status messages
    -t TIMEOUT | --timeout=TIMEOUT
                                Timeout in seconds, zero for no timeout
    -- COMMAND ARGS             Execute command with args after the test finishes
USAGE
    exit 1
}

wait_for()
{
    if [[ $TIMEOUT -gt 0 ]]; then
        echoerr "$cmdname: waiting $TIMEOUT seconds for $HOST:$PORT"
    else
        echoerr "$cmdname: waiting for $HOST:$PORT without a timeout"
    fi
    start_ts=$(date +%s)
    while :
    do
        if [[ $ISBUSY -eq 1 ]]; then
            nc -z $HOST $PORT
            result=$?
        else
            (echo > /dev/tcp/$HOST/$PORT) >/dev/null 2>&1
            result=$?
        fi
        if [[ $result -eq 0 ]]; then
            end_ts=$(date +%s)
            echoerr "$cmdname: $HOST:$PORT is available after $((end_ts - start_ts)) seconds"
            break
        fi
        sleep 1
    done
    return $result
}

wait_for_wrapper()
{
    # In order to support SIGINT during timeout: http://unix.stackexchange.com/a/57692
    if [[ $QUIET -eq 1 ]]; then
        timeout $BUSYTIMEFLAG $TIMEOUT $0 --quiet --child --host=$HOST --port=$PORT --timeout=$TIMEOUT &
    else
        timeout $BUSYTIMEFLAG $TIMEOUT $0 --child --host=$HOST --port=$PORT --timeout=$TIMEOUT &
    fi
    PID=$!
    trap "kill -INT -$PID" INT
    wait $PID
    RESULT=$?
    if [[ $RESULT -ne 0 ]]; then
        echoerr "$cmdname: timeout occurred after waiting $TIMEOUT seconds for $HOST:$PORT"
    fi
    return $RESULT
}

# process arguments
while [[ $# -gt 0 ]]
do
    case "$1" in
        *:* )
        hostport=(${1//:/ })
        HOST=${hostport[0]}
        PORT=${hostport[1]}
        shift 1
        ;;
        --child)
        CHILD=1
        shift 1
        ;;
        -q | --quiet)
        QUIET=1
        shift 1
        ;;
        -s | --strict)
        STRICT=1
        shift 1
        ;;
        -h)
        HOST="$2"
        if [[ $HOST == "" ]]; then break; fi
        shift 2
        ;;
        --host=*)
        HOST="${1#*=}"
        shift 1
        ;;
        -p)
        PORT="$2"
        if [[ $PORT == "" ]]; then break; fi
        shift 2
        ;;
        --port=*)
        PORT="${1#*=}"
        shift 1
        ;;
        -t)
        TIMEOUT="$2"
        if [[ $TIMEOUT == "" ]]; then break; fi
        shift 2
        ;;
        --timeout=*)
        TIMEOUT="${1#*=}"
        shift 1
        ;;
        --)
        shift
        CLI=("$@")
        break
        ;;
        --help)
        usage
        ;;
        *)
        echoerr "Unknown argument: $1"
        usage
        ;;
    esac
done

if [[ "$HOST" == "" || "$PORT" == "" ]]; then
    echoerr "Error: you need to provide a host and port to test."
    usage
fi

TIMEOUT=${TIMEOUT:-15}
STRICT=${STRICT:-0}
CHILD=${CHILD:-0}
QUIET=${QUIET:-0}

# check to see if timeout is from busybox?
# check to see if timeout is from busybox?
TIMEOUT_PATH=$(realpath $(which timeout))
if [[ $TIMEOUT_PATH =~ "busybox" ]]; then
        ISBUSY=1
        BUSYTIMEFLAG="-t"
else
        ISBUSY=0
        BUSYTIMEFLAG=""
fi

if [[ $CHILD -gt 0 ]]; then
    wait_for
    RESULT=$?
    exit $RESULT
else
    if [[ $TIMEOUT -gt 0 ]]; then
        wait_for_wrapper
        RESULT=$?
    else
        wait_for
        RESULT=$?
    fi
fi

if [[ $CLI != "" ]]; then
    if [[ $RESULT -ne 0 && $STRICT -eq 1 ]]; then
        echoerr "$cmdname: strict mode, refusing to execute subprocess"
        exit $RESULT
    fi
    exec "${CLI[@]}"
else
    exit $RESULT
fi

运行服务器.sh

#! /bin/sh

./wait-for-it.sh $PG_HOST:$PG_PORT -t 60

./wait-for-it.sh $ELASTICSEARCH_HOST:$ELASTICSEARCH_PORT -t 60

./wait-for-it.sh $REDIS_HOST:$REDIS_PORT -t 60

bootstrap -v

pgsync -v --daemon

**错误消息:**

pgsync-1  | Traceback (most recent call last):
pgsync-1  |   File "/usr/local/bin/pgsync", line 7, in <module>
pgsync-1  |  0:00:00.202205 (0.20 sec)
pgsync-1  |     sync.main()
pgsync-1  |   File "/usr/local/lib/python3.11/site-packages/click/core.py", line 1157, in __call__
pgsync-1  |     return self.main(*args, **kwargs)
pgsync-1  |            ^^^^^^^^^^^^^^^^^^^^^^^^^^
pgsync-1  |   File "/usr/local/lib/python3.11/site-packages/click/core.py", line 1078, in main
pgsync-1  |     rv = self.invoke(ctx)
pgsync-1  |          ^^^^^^^^^^^^^^^^
pgsync-1  |   File "/usr/local/lib/python3.11/site-packages/click/core.py", line 1434, in invoke
pgsync-1  |     return ctx.invoke(self.callback, **ctx.params)
pgsync-1  |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
pgsync-1  |   File "/usr/local/lib/python3.11/site-packages/click/core.py", line 783, in invoke
pgsync-1  |     return __callback(*args, **kwargs)
pgsync-1  |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^
pgsync-1  |   File "/usr/local/lib/python3.11/site-packages/pgsync/sync.py", line 1480, in main
pgsync-1  |     sync: Sync = Sync(
pgsync-1  |                  ^^^^^
pgsync-1  |   File "/usr/local/lib/python3.11/site-packages/pgsync/singleton.py", line 36, in __call__
pgsync-1  |     cls._instances[key] = super(Singleton, cls).__call__(
pgsync-1  |                           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
pgsync-1  |   File "/usr/local/lib/python3.11/site-packages/pgsync/sync.py", line 88, in __init__
pgsync-1  |     self.search_client: SearchClient = SearchClient()
pgsync-1  |                                        ^^^^^^^^^^^^^^
pgsync-1  |   File "/usr/local/lib/python3.11/site-packages/pgsync/search_client.py", line 48, in __init__
pgsync-1  |     self.__client.info()["version"]["number"].split(".")[0]
pgsync-1  |     ^^^^^^^^^^^^^^^^^^^^
pgsync-1  |   File "/usr/local/lib/python3.11/site-packages/elasticsearch/_sync/client/utils.py", line 402, in wrapped
pgsync-1  |     return api(*args, **kwargs)
pgsync-1  |            ^^^^^^^^^^^^^^^^^^^^
pgsync-1  |   File "/usr/local/lib/python3.11/site-packages/elasticsearch/_sync/client/__init__.py", line 2278, in info
pgsync-1  |     return self.perform_request(  # type: ignore[return-value]
pgsync-1  |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
pgsync-1  |   File "/usr/local/lib/python3.11/site-packages/elasticsearch/_sync/client/_base.py", line 320, in perform_request
pgsync-1  |     raise HTTP_EXCEPTIONS.get(meta.status, ApiError)(
pgsync-1  | elasticsearch.AuthenticationException: AuthenticationException(401, 'security_exception', 'unable to authenticate user [elastic] for REST request [/]')

我试过: -已检查的密码环境变量可用于pgsync -检查过我可以通过kibana和curl登录。我做得很成功。(虽然我还没有测试提供卷曲的弹性搜索证书) -关闭pgsync上的SSL、Schema和证书验证环境变量可以使身份验证再次工作。尽管它超出了我的目的。