From 97d182d2e7199b528c472e778aaf469e4e07fb8e Mon Sep 17 00:00:00 2001 From: CodersKin Date: Wed, 10 Jun 2026 13:50:35 +0300 Subject: [PATCH 01/16] Fix: included IMPORT_KB_ON_START as default to be 0 for autotests --- .github/workflows/common.yml | 1 + scripts/omegaclaw | 1 + 2 files changed, 2 insertions(+) diff --git a/.github/workflows/common.yml b/.github/workflows/common.yml index 439ca209..bd32004b 100644 --- a/.github/workflows/common.yml +++ b/.github/workflows/common.yml @@ -101,6 +101,7 @@ jobs: fi env: TEST_SERVER_IP: 'host.docker.internal' + IMPORT_KB_ON_START: '0' - name: Autotests - Phase 1 (Blocking Checks) if: inputs.run_tests diff --git a/scripts/omegaclaw b/scripts/omegaclaw index bb960349..bbd43d6d 100755 --- a/scripts/omegaclaw +++ b/scripts/omegaclaw @@ -496,6 +496,7 @@ start() { -e "SL_BOT_TOKEN=${SL_BOT_TOKEN:-}" -e OMEGACLAW_AUTH_SECRET="$OMEGACLAW_AUTH_SECRET" -e LLM_SERVER_LOCAL_URL="$llm_server_local_url" + -e IMPORT_KB_ON_START="${IMPORT_KB_ON_START:-1}" "$image" "commchannel=${commchannel}" "provider=${provider}" From 23aa06888d992842b1608616ee5cb106555e406a Mon Sep 17 00:00:00 2001 From: CodersKin Date: Wed, 10 Jun 2026 17:40:38 +0300 Subject: [PATCH 02/16] Chore: kept entrypoint and import-knowledge script separate, added a small doc for running import-kb script without docker, reverted lib_llb_ext to core version, cleaned DockerFile --- Dockerfile | 5 ++-- README.md | 9 ++++++ entrypoint.sh | 67 +++++---------------------------------------- import_knowledge.sh | 57 ++++++++++++++++++++++++++++++++++++++ lib_llm_ext.py | 2 +- 5 files changed, 76 insertions(+), 64 deletions(-) create mode 100644 import_knowledge.sh diff --git a/Dockerfile b/Dockerfile index f57875d6..6aaa8b58 100644 --- a/Dockerfile +++ b/Dockerfile @@ -105,14 +105,12 @@ RUN mkdir /opt/nginx RUN chown www-data:www-data /opt/nginx RUN chmod 0700 /opt/nginx COPY --chown=www-data:www-data --chmod=0600 ./proxy/* /opt/nginx/ -RUN chmod 0600 /opt/nginx/* ENV OMEGACLAW_DIR=/PeTTa/repos/OmegaClaw-Core ENV MEMORY_DIR=${OMEGACLAW_DIR}/memory # Start defaults for import-kb -ENV CHROMA_DB_PATH=/PeTTa/chroma_db ENV IMPORT_KB_ON_START=1 -ENV EMBEDDING_PROVIDER=OpenAI +ENV EMBEDDING_MODEL=${EMBEDDING_MODEL} # Bring in only local OmegaClaw source (filtered by .dockerignore). COPY . ${OMEGACLAW_DIR} @@ -121,6 +119,7 @@ RUN cp ${OMEGACLAW_DIR}/run.metta /PeTTa/run.metta \ && mkdir -p ${MEMORY_DIR}/chroma_db \ && ln -s ${MEMORY_DIR}/chroma_db ./chroma_db \ && chmod +x ${OMEGACLAW_DIR}/entrypoint.sh \ + && chmod +x ${OMEGACLAW_DIR}/import-knowledge.sh \ && chown -R 65534:65534 ${MEMORY_DIR} \ && find ${MEMORY_DIR} -type f -exec chmod 0644 {} \; \ && chmod 0444 ${MEMORY_DIR}/prompt.txt \ diff --git a/README.md b/README.md index 5a82e9b1..3831a4f7 100644 --- a/README.md +++ b/README.md @@ -82,6 +82,15 @@ OMEGACLAW_AUTH_SECRET= sh run.sh run.metta IRC_channel="` and after agent is joined send `auth ` message to authenticate yourself as an agent owner. Please replace `` and `` by your own values. +### Import-knowledge + +If your running OmegaClaw without the docker and would like to load it with a preset Knowledge you can do so by running the following: + +1. Set `embeddingprovider` in your env, Note that it can be OpenAI or Local; OpenAI embeddings will require `OpenAI_API_KEY` in your env as well +2. Run `sh ./import_knowledge.sh` + +After the script is done your OmegaClaw bot will have the preset knowledge in its LTM. + ## Reference — Configuration Options ### General diff --git a/entrypoint.sh b/entrypoint.sh index a7f11cbb..7d358c78 100644 --- a/entrypoint.sh +++ b/entrypoint.sh @@ -3,15 +3,12 @@ set -euo pipefail cd /PeTTa -# 1. Start Nginx su www-data -s /bin/sh -c "sh /opt/nginx/nginx.sh" -# 2. Setup Database Variables export CHROMA_DB_PATH="${CHROMA_DB_PATH:-/PeTTa/chroma_db}" -IMPORT_KB_ON_START="${IMPORT_KB_ON_START:-1}" -IMPORT_KB_FORCE="${IMPORT_KB_FORCE:-0}" -EMBEDDING_PROVIDER="${embeddingprovider:-Local}" -GATEWAY_URL="http://localhost:8080" +export IMPORT_KB_ON_START="${IMPORT_KB_ON_START:-1}" +export EMBEDDING_PROVIDER="${EMBEDDING_PROVIDER:-${embeddingprovider:-Local}}" +export GATEWAY_URL="http://localhost:8080" for arg in "$@"; do if [[ "$arg" == embeddingprovider=* ]]; then @@ -21,65 +18,16 @@ done mkdir -p "${CHROMA_DB_PATH}" -normalize_provider() { - echo "$1" | tr '[:upper:]' '[:lower:]' -} - -# 3. Handle Knowledge Base Import +# Optional knowledge-base import if [[ "${IMPORT_KB_ON_START}" == "1" ]]; then - PROVIDER="$(normalize_provider "${EMBEDDING_PROVIDER}")" - - case "${PROVIDER}" in - openai) - if [[ -z "${OPENAI_API_KEY:-}" ]]; then - echo "ERROR: OPENAI_API_KEY is required when EMBEDDING_PROVIDER=OpenAI." >&2 - exit 1 - fi - - SENTINEL="${CHROMA_DB_PATH}/.import-kb.openai.done" - - if [[ -f "${SENTINEL}" && "${IMPORT_KB_FORCE}" != "1" ]]; then - echo "[entrypoint] import-kb already initialized with OpenAI embeddings; skipping." - else - echo "[entrypoint] Running import-kb with default OpenAI embeddings." - echo "[entrypoint] CHROMA_DB_PATH=${CHROMA_DB_PATH}" - import-knowledge - date -Iseconds > "${SENTINEL}" - echo "[entrypoint] import-kb complete." - fi - ;; - - local) - SENTINEL="${CHROMA_DB_PATH}/.import-kb.local.done" - - if [[ -f "${SENTINEL}" && "${IMPORT_KB_FORCE}" != "1" ]]; then - echo "[entrypoint] import-kb already initialized with local embeddings; skipping." - else - echo "[entrypoint] Running import-kb with default local embeddings." - echo "[entrypoint] CHROMA_DB_PATH=${CHROMA_DB_PATH}" - import-knowledge --local - date -Iseconds > "${SENTINEL}" - echo "[entrypoint] import-kb complete." - fi - ;; - - *) - echo "ERROR: Unsupported EMBEDDING_PROVIDER='${EMBEDDING_PROVIDER}'." >&2 - echo "Use EMBEDDING_PROVIDER=OpenAI or EMBEDDING_PROVIDER=Local." >&2 - exit 1 - ;; - esac - - # Ensure the runtime user 'nobody' has permissions to read/write the db - chown -R nobody "${CHROMA_DB_PATH}" 2>/dev/null || true + "${OMEGACLAW_DIR}/import-knowledge.sh" fi -# 4. Scrub environment: only allowlisted vars survive. +# Scrub environment: only allowlisted vars survive. SAFE_VARS="HOME USER PATH HOSTNAME TERM LANG LC_ALL \ GATEWAY_URL PYTHONDONTWRITEBYTECODE PYTHONUNBUFFERED \ HF_HOME SENTENCE_TRANSFORMERS_HOME HF_HUB_OFFLINE TRANSFORMERS_OFFLINE \ - OMEGACLAW_DIR MEMORY_DIR LLM_SERVER_LOCAL_URL TEST_SERVER_IP \ - CHROMA_DB_PATH OPENAI_API_KEY embeddingprovider" + OMEGACLAW_DIR MEMORY_DIR LLM_SERVER_LOCAL_URL TEST_SERVER_IP" env_args="" for var in $SAFE_VARS; do @@ -89,5 +37,4 @@ for var in $SAFE_VARS; do fi done -# 5. Execute core application exec env -i $env_args su nobody -s /bin/sh -c "sh run.sh run.metta $*" \ No newline at end of file diff --git a/import_knowledge.sh b/import_knowledge.sh new file mode 100644 index 00000000..ac96b779 --- /dev/null +++ b/import_knowledge.sh @@ -0,0 +1,57 @@ +#!/bin/sh +set -euo pipefail + +export CHROMA_DB_PATH="${CHROMA_DB_PATH:-/PeTTa/chroma_db}" +IMPORT_KB_FORCE="${IMPORT_KB_FORCE:-0}" +EMBEDDING_PROVIDER="${EMBEDDING_PROVIDER:-${embeddingprovider:-Local}}" + +normalize_provider() { + echo "$1" | tr '[:upper:]' '[:lower:]' +} + +mkdir -p "${CHROMA_DB_PATH}" + +PROVIDER="$(normalize_provider "${EMBEDDING_PROVIDER}")" + +case "${PROVIDER}" in + openai) + if [[ -z "${OPENAI_API_KEY:-}" ]]; then + echo "ERROR: OPENAI_API_KEY is required when EMBEDDING_PROVIDER=OpenAI." >&2 + exit 1 + fi + + SENTINEL="${CHROMA_DB_PATH}/.import-kb.openai.done" + + if [[ -f "${SENTINEL}" && "${IMPORT_KB_FORCE}" != "1" ]]; then + echo "[import-kb] Already initialized with OpenAI embeddings; skipping." + else + echo "[import-kb] Running import-knowledge with OpenAI embeddings." + echo "[import-kb] CHROMA_DB_PATH=${CHROMA_DB_PATH}" + import-knowledge + date -Iseconds > "${SENTINEL}" + echo "[import-kb] Import complete." + fi + ;; + + local) + SENTINEL="${CHROMA_DB_PATH}/.import-kb.local.done" + + if [[ -f "${SENTINEL}" && "${IMPORT_KB_FORCE}" != "1" ]]; then + echo "[import-kb] Already initialized with local embeddings; skipping." + else + echo "[import-kb] Running import-knowledge with local embeddings." + echo "[import-kb] CHROMA_DB_PATH=${CHROMA_DB_PATH}" + import-knowledge --local --model ${EMBEDDING_MODEL:-intfloat/e5-large-v2} + date -Iseconds > "${SENTINEL}" + echo "[import-kb] Import complete." + fi + ;; + + *) + echo "ERROR: Unsupported EMBEDDING_PROVIDER='${EMBEDDING_PROVIDER}'." >&2 + echo "Use EMBEDDING_PROVIDER=OpenAI or EMBEDDING_PROVIDER=Local." >&2 + exit 1 + ;; +esac + +chown -R nobody "${CHROMA_DB_PATH}" 2>/dev/null || true \ No newline at end of file diff --git a/lib_llm_ext.py b/lib_llm_ext.py index 44a0578a..104d3103 100644 --- a/lib_llm_ext.py +++ b/lib_llm_ext.py @@ -216,7 +216,7 @@ def _get_provider(name: str) -> Optional[AIProvider]: _register_provider(name="Anthropic", var_name="ANTHROPIC_API_KEY", model_name="claude-opus-4-6", base_url="https://api.anthropic.com/v1/") _register_provider(name="Ollama-local", var_name="OLLAMA_API_KEY", model_name="qwen3.5:9b", base_url="http://localhost:11434/v1") _register_provider_instance(AsiOneProvider(name="ASIOne", var_name="ASIONE_API_KEY", model_name="asi1-ultra", base_url="https://api.asi1.ai/v1")) -_register_provider(name="OpenRouter", var_name="OPENROUTER_API_KEY", model_name="z-ai/glm-5.1", base_url="https://openrouter.ai/api/v1") +_register_provider_instance(OpenRouterProvider(name="OpenRouter", var_name="OPENROUTER_API_KEY", model_name="z-ai/glm-5.1", base_url="https://openrouter.ai/api/v1")) _register_provider_instance(TestProvider()) _register_provider_instance(OpenAIProvider(name="OpenAI", var_name="OPENAI_API_KEY", model_name="gpt-5.4", base_url="https://api.openai.com/v1")) From 5906729ecdadbbd25b66312556a0892d9b49ad2a Mon Sep 17 00:00:00 2001 From: CodersKin Date: Wed, 10 Jun 2026 18:32:35 +0300 Subject: [PATCH 03/16] Fix: fixed some typo errors --- Dockerfile | 2 +- entrypoint.sh | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/Dockerfile b/Dockerfile index 6aaa8b58..01a8abfe 100644 --- a/Dockerfile +++ b/Dockerfile @@ -119,7 +119,7 @@ RUN cp ${OMEGACLAW_DIR}/run.metta /PeTTa/run.metta \ && mkdir -p ${MEMORY_DIR}/chroma_db \ && ln -s ${MEMORY_DIR}/chroma_db ./chroma_db \ && chmod +x ${OMEGACLAW_DIR}/entrypoint.sh \ - && chmod +x ${OMEGACLAW_DIR}/import-knowledge.sh \ + && chmod +x ${OMEGACLAW_DIR}/import_knowledge.sh \ && chown -R 65534:65534 ${MEMORY_DIR} \ && find ${MEMORY_DIR} -type f -exec chmod 0644 {} \; \ && chmod 0444 ${MEMORY_DIR}/prompt.txt \ diff --git a/entrypoint.sh b/entrypoint.sh index 7d358c78..a13ef7bb 100644 --- a/entrypoint.sh +++ b/entrypoint.sh @@ -20,7 +20,7 @@ mkdir -p "${CHROMA_DB_PATH}" # Optional knowledge-base import if [[ "${IMPORT_KB_ON_START}" == "1" ]]; then - "${OMEGACLAW_DIR}/import-knowledge.sh" + "${OMEGACLAW_DIR}/import_knowledge.sh" fi # Scrub environment: only allowlisted vars survive. From 2b66c73b8b7cbe7719488f383cc36b837e9fe850 Mon Sep 17 00:00:00 2001 From: CodersKin Date: Wed, 10 Jun 2026 19:06:32 +0300 Subject: [PATCH 04/16] chore: moved import-kb script into scripts folder --- entrypoint.sh | 2 +- import_knowledge.sh => scripts/import_knowledge.sh | 0 2 files changed, 1 insertion(+), 1 deletion(-) rename import_knowledge.sh => scripts/import_knowledge.sh (100%) diff --git a/entrypoint.sh b/entrypoint.sh index a13ef7bb..5bfe2914 100644 --- a/entrypoint.sh +++ b/entrypoint.sh @@ -20,7 +20,7 @@ mkdir -p "${CHROMA_DB_PATH}" # Optional knowledge-base import if [[ "${IMPORT_KB_ON_START}" == "1" ]]; then - "${OMEGACLAW_DIR}/import_knowledge.sh" + "${OMEGACLAW_DIR}/scripts/import_knowledge.sh" fi # Scrub environment: only allowlisted vars survive. diff --git a/import_knowledge.sh b/scripts/import_knowledge.sh similarity index 100% rename from import_knowledge.sh rename to scripts/import_knowledge.sh From 93cb43ade8f7ef9ca807690edba6283bd10b2bc2 Mon Sep 17 00:00:00 2001 From: CodersKin Date: Wed, 10 Jun 2026 19:35:34 +0300 Subject: [PATCH 05/16] Fix: fixed some typo error --- Dockerfile | 2 +- scripts/import_knowledge.sh | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/Dockerfile b/Dockerfile index 01a8abfe..922ca5c6 100644 --- a/Dockerfile +++ b/Dockerfile @@ -119,7 +119,7 @@ RUN cp ${OMEGACLAW_DIR}/run.metta /PeTTa/run.metta \ && mkdir -p ${MEMORY_DIR}/chroma_db \ && ln -s ${MEMORY_DIR}/chroma_db ./chroma_db \ && chmod +x ${OMEGACLAW_DIR}/entrypoint.sh \ - && chmod +x ${OMEGACLAW_DIR}/import_knowledge.sh \ + && chmod +x ${OMEGACLAW_DIR}/scripts/import_knowledge.sh \ && chown -R 65534:65534 ${MEMORY_DIR} \ && find ${MEMORY_DIR} -type f -exec chmod 0644 {} \; \ && chmod 0444 ${MEMORY_DIR}/prompt.txt \ diff --git a/scripts/import_knowledge.sh b/scripts/import_knowledge.sh index ac96b779..7951247c 100644 --- a/scripts/import_knowledge.sh +++ b/scripts/import_knowledge.sh @@ -1,4 +1,4 @@ -#!/bin/sh +#!/usr/bin/env bash set -euo pipefail export CHROMA_DB_PATH="${CHROMA_DB_PATH:-/PeTTa/chroma_db}" From a3cc4d2f959a8cf331df81de54679512666c328b Mon Sep 17 00:00:00 2001 From: CodersKin Date: Thu, 11 Jun 2026 13:15:31 +0300 Subject: [PATCH 06/16] Chore: cleaned up the code , added comments and improved README --- .github/workflows/common.yml | 2 +- Dockerfile | 1 - README.md | 15 ++++++++++----- entrypoint.sh | 15 ++------------- scripts/import_knowledge.sh | 16 ++++++++++------ 5 files changed, 23 insertions(+), 26 deletions(-) diff --git a/.github/workflows/common.yml b/.github/workflows/common.yml index bd32004b..97e2a9ec 100644 --- a/.github/workflows/common.yml +++ b/.github/workflows/common.yml @@ -101,7 +101,7 @@ jobs: fi env: TEST_SERVER_IP: 'host.docker.internal' - IMPORT_KB_ON_START: '0' + IMPORT_KB_ON_START: '0' # TODO: this is turned off because Python dependencies are not installed to the workflow environment - name: Autotests - Phase 1 (Blocking Checks) if: inputs.run_tests diff --git a/Dockerfile b/Dockerfile index 922ca5c6..8a6e0769 100644 --- a/Dockerfile +++ b/Dockerfile @@ -110,7 +110,6 @@ ENV OMEGACLAW_DIR=/PeTTa/repos/OmegaClaw-Core ENV MEMORY_DIR=${OMEGACLAW_DIR}/memory # Start defaults for import-kb ENV IMPORT_KB_ON_START=1 -ENV EMBEDDING_MODEL=${EMBEDDING_MODEL} # Bring in only local OmegaClaw source (filtered by .dockerignore). COPY . ${OMEGACLAW_DIR} diff --git a/README.md b/README.md index 3831a4f7..05d4fe3a 100644 --- a/README.md +++ b/README.md @@ -82,14 +82,19 @@ OMEGACLAW_AUTH_SECRET= sh run.sh run.metta IRC_channel="` and after agent is joined send `auth ` message to authenticate yourself as an agent owner. Please replace `` and `` by your own values. -### Import-knowledge +### Import Knowledge -If your running OmegaClaw without the docker and would like to load it with a preset Knowledge you can do so by running the following: +If you are running OmegaClaw without Docker and would like to load it with preset knowledge, follow these steps: -1. Set `embeddingprovider` in your env, Note that it can be OpenAI or Local; OpenAI embeddings will require `OpenAI_API_KEY` in your env as well -2. Run `sh ./import_knowledge.sh` +1. Set embeddingprovider in your environment. It can be set to either OpenAI or Local. OpenAI embeddings also require OPENAI_API_KEY to be set in your environment. -After the script is done your OmegaClaw bot will have the preset knowledge in its LTM. +2. Run: +``` + sh ./import_knowledge.sh + After the script finishes, your OmegaClaw bot will have the preset knowledge stored in its long-term memory (LTM). +``` + +If you want to skip preloading the knowledge then run `export IMPORT_KB_ON_START=0` ## Reference — Configuration Options diff --git a/entrypoint.sh b/entrypoint.sh index 5bfe2914..af76c113 100644 --- a/entrypoint.sh +++ b/entrypoint.sh @@ -5,22 +5,11 @@ cd /PeTTa su www-data -s /bin/sh -c "sh /opt/nginx/nginx.sh" -export CHROMA_DB_PATH="${CHROMA_DB_PATH:-/PeTTa/chroma_db}" -export IMPORT_KB_ON_START="${IMPORT_KB_ON_START:-1}" -export EMBEDDING_PROVIDER="${EMBEDDING_PROVIDER:-${embeddingprovider:-Local}}" -export GATEWAY_URL="http://localhost:8080" - -for arg in "$@"; do - if [[ "$arg" == embeddingprovider=* ]]; then - export EMBEDDING_PROVIDER="${arg#*=}" - fi -done - -mkdir -p "${CHROMA_DB_PATH}" +GATEWAY_URL="http://localhost:8080" # Optional knowledge-base import if [[ "${IMPORT_KB_ON_START}" == "1" ]]; then - "${OMEGACLAW_DIR}/scripts/import_knowledge.sh" + su nobody -s /bin/sh -c "${OMEGACLAW_DIR}/scripts/import_knowledge.sh" fi # Scrub environment: only allowlisted vars survive. diff --git a/scripts/import_knowledge.sh b/scripts/import_knowledge.sh index 7951247c..531c6575 100644 --- a/scripts/import_knowledge.sh +++ b/scripts/import_knowledge.sh @@ -1,9 +1,15 @@ #!/usr/bin/env bash set -euo pipefail -export CHROMA_DB_PATH="${CHROMA_DB_PATH:-/PeTTa/chroma_db}" +CHROMA_DB_PATH="${CHROMA_DB_PATH:-/PeTTa/chroma_db}" IMPORT_KB_FORCE="${IMPORT_KB_FORCE:-0}" -EMBEDDING_PROVIDER="${EMBEDDING_PROVIDER:-${embeddingprovider:-Local}}" +EMBEDDING_PROVIDER="${embeddingprovider:-Local}" + +for arg in "$@"; do + if [[ "$arg" == embeddingprovider=* ]]; then + export EMBEDDING_PROVIDER="${arg#*=}" + fi +done normalize_provider() { echo "$1" | tr '[:upper:]' '[:lower:]' @@ -41,7 +47,7 @@ case "${PROVIDER}" in else echo "[import-kb] Running import-knowledge with local embeddings." echo "[import-kb] CHROMA_DB_PATH=${CHROMA_DB_PATH}" - import-knowledge --local --model ${EMBEDDING_MODEL:-intfloat/e5-large-v2} + import-knowledge --local date -Iseconds > "${SENTINEL}" echo "[import-kb] Import complete." fi @@ -52,6 +58,4 @@ case "${PROVIDER}" in echo "Use EMBEDDING_PROVIDER=OpenAI or EMBEDDING_PROVIDER=Local." >&2 exit 1 ;; -esac - -chown -R nobody "${CHROMA_DB_PATH}" 2>/dev/null || true \ No newline at end of file +esac \ No newline at end of file From e1dba09c03b25918b386886bbd9a5de2548ff60b Mon Sep 17 00:00:00 2001 From: CodersKin Date: Thu, 11 Jun 2026 14:00:44 +0300 Subject: [PATCH 07/16] Fix: moved EMBEDDING_PROVIDER def into entrypoint because it couldn't find the argument embeddingprovider in import-kb script --- entrypoint.sh | 7 +++++++ scripts/import_knowledge.sh | 11 ++--------- 2 files changed, 9 insertions(+), 9 deletions(-) diff --git a/entrypoint.sh b/entrypoint.sh index af76c113..dac745b3 100644 --- a/entrypoint.sh +++ b/entrypoint.sh @@ -6,6 +6,13 @@ cd /PeTTa su www-data -s /bin/sh -c "sh /opt/nginx/nginx.sh" GATEWAY_URL="http://localhost:8080" +EMBEDDING_PROVIDER="${embeddingprovider:-Local}" + +for arg in "$@"; do + if [[ "$arg" == embeddingprovider=* ]]; then + export EMBEDDING_PROVIDER="${arg#*=}" + fi +done # Optional knowledge-base import if [[ "${IMPORT_KB_ON_START}" == "1" ]]; then diff --git a/scripts/import_knowledge.sh b/scripts/import_knowledge.sh index 531c6575..33cb0777 100644 --- a/scripts/import_knowledge.sh +++ b/scripts/import_knowledge.sh @@ -3,13 +3,6 @@ set -euo pipefail CHROMA_DB_PATH="${CHROMA_DB_PATH:-/PeTTa/chroma_db}" IMPORT_KB_FORCE="${IMPORT_KB_FORCE:-0}" -EMBEDDING_PROVIDER="${embeddingprovider:-Local}" - -for arg in "$@"; do - if [[ "$arg" == embeddingprovider=* ]]; then - export EMBEDDING_PROVIDER="${arg#*=}" - fi -done normalize_provider() { echo "$1" | tr '[:upper:]' '[:lower:]' @@ -54,8 +47,8 @@ case "${PROVIDER}" in ;; *) - echo "ERROR: Unsupported EMBEDDING_PROVIDER='${EMBEDDING_PROVIDER}'." >&2 - echo "Use EMBEDDING_PROVIDER=OpenAI or EMBEDDING_PROVIDER=Local." >&2 + echo "ERROR: Unsupported embeddingprovider='${EMBEDDING_PROVIDER}'." >&2 + echo "Use embeddingprovider=OpenAI or embeddingprovider=Local." >&2 exit 1 ;; esac \ No newline at end of file From 391cd6455aab31692ef89c514082e27af616b195 Mon Sep 17 00:00:00 2001 From: CodersKin Date: Thu, 11 Jun 2026 16:13:16 +0300 Subject: [PATCH 08/16] chore: removed white space --- Dockerfile | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/Dockerfile b/Dockerfile index 8a6e0769..3b80e273 100644 --- a/Dockerfile +++ b/Dockerfile @@ -79,17 +79,17 @@ ENV DEBIAN_FRONTEND=noninteractive \ RUN apt-get update \ && apt-get install -y --no-install-recommends \ - ca-certificates \ - python3 \ - libopenblas-dev \ - libblas-dev \ - liblapack-dev \ - gfortran \ - libgflags-dev \ - nano \ - git \ - nginx-light \ - gettext-base \ + ca-certificates \ + python3 \ + libopenblas-dev \ + libblas-dev \ + liblapack-dev \ + gfortran \ + libgflags-dev \ + nano \ + git \ + nginx-light \ + gettext-base \ && rm -rf /var/lib/apt/lists/* WORKDIR /PeTTa From 98416faf32c57dc867d1200b7eef4ba30b5263da Mon Sep 17 00:00:00 2001 From: CodersKin Date: Thu, 11 Jun 2026 16:22:37 +0300 Subject: [PATCH 09/16] chore: removed whitespace --- Dockerfile | 50 +++++++++++++++++++++++++------------------------- 1 file changed, 25 insertions(+), 25 deletions(-) diff --git a/Dockerfile b/Dockerfile index 3b80e273..28e36f1f 100644 --- a/Dockerfile +++ b/Dockerfile @@ -12,20 +12,20 @@ ENV DEBIAN_FRONTEND=noninteractive \ RUN apt-get update \ && apt-get install -y --no-install-recommends \ - ca-certificates \ - git \ - build-essential \ - cmake \ - pkg-config \ - python3 \ - python3-dev \ - python3-pip \ - libopenblas-dev \ - libblas-dev \ - liblapack-dev \ - gfortran \ - libgflags-dev \ - nano \ + ca-certificates \ + git \ + build-essential \ + cmake \ + pkg-config \ + python3 \ + python3-dev \ + python3-pip \ + libopenblas-dev \ + libblas-dev \ + liblapack-dev \ + gfortran \ + libgflags-dev \ + nano \ && rm -rf /var/lib/apt/lists/* # Build dependencies from source. Pin refs at build time for reproducibility. @@ -79,17 +79,17 @@ ENV DEBIAN_FRONTEND=noninteractive \ RUN apt-get update \ && apt-get install -y --no-install-recommends \ - ca-certificates \ - python3 \ - libopenblas-dev \ - libblas-dev \ - liblapack-dev \ - gfortran \ - libgflags-dev \ - nano \ - git \ - nginx-light \ - gettext-base \ + ca-certificates \ + python3 \ + libopenblas-dev \ + libblas-dev \ + liblapack-dev \ + gfortran \ + libgflags-dev \ + nano \ + git \ + nginx-light \ + gettext-base \ && rm -rf /var/lib/apt/lists/* WORKDIR /PeTTa From 65a54a226d3e48a0de7097ccf391e6bd70a2cfab Mon Sep 17 00:00:00 2001 From: CodersKin Date: Thu, 11 Jun 2026 17:41:02 +0300 Subject: [PATCH 10/16] Feat: added options inside omegaclaw script --- scripts/omegaclaw | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/scripts/omegaclaw b/scripts/omegaclaw index bbd43d6d..e0cc23ef 100755 --- a/scripts/omegaclaw +++ b/scripts/omegaclaw @@ -235,6 +235,24 @@ def _choose_provider(): return provider, embeddingprovider, api_token_var, llm_server_local_url +def _choose_import_kb(): + while True: + c = input( + "Import knowledge base on startup? [Y/n]: " + ).strip().lower() + + if c in ("", "y", "yes"): + return "1" + + if c in ("n", "no"): + return "0" + + if c == "q": + sys.exit(1) + + print("Please enter y or n.") + + def _prompt_llm_token(): while True: token = getpass.getpass("Please paste your LLM token and press ENTER or 'q' to exit: ").strip() @@ -261,6 +279,7 @@ def config_run_omegaclaw(config_output_path): channel_config = _choose_channel() provider, embeddingprovider, api_token_var, llm_server_local_url = _choose_provider() + import_kb_on_start = _choose_import_kb() token = _prompt_llm_token() with open(config_output_path, "w", encoding="utf-8") as f: @@ -270,6 +289,7 @@ def config_run_omegaclaw(config_output_path): _write_kv(f, "provider", provider) _write_kv(f, "embeddingprovider", embeddingprovider) _write_kv(f, "llm_server_local_url", llm_server_local_url) + _write_kv(f, "IMPORT_KB_ON_START", import_kb_on_start) for key, value in channel_config.items(): _write_kv(f, key, value) From 1844865d283187c2e527518257ad0e5a99e097b9 Mon Sep 17 00:00:00 2001 From: CodersKin Date: Thu, 11 Jun 2026 19:17:30 +0300 Subject: [PATCH 11/16] Fix: fixed some typo in README and made import-kb prompt more readable --- README.md | 2 +- scripts/omegaclaw | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 05d4fe3a..7fca6c9e 100644 --- a/README.md +++ b/README.md @@ -91,8 +91,8 @@ If you are running OmegaClaw without Docker and would like to load it with prese 2. Run: ``` sh ./import_knowledge.sh - After the script finishes, your OmegaClaw bot will have the preset knowledge stored in its long-term memory (LTM). ``` +After the script finishes, your OmegaClaw bot will have the preset knowledge stored in its long-term memory (LTM). If you want to skip preloading the knowledge then run `export IMPORT_KB_ON_START=0` diff --git a/scripts/omegaclaw b/scripts/omegaclaw index e0cc23ef..5fa18fca 100755 --- a/scripts/omegaclaw +++ b/scripts/omegaclaw @@ -238,7 +238,7 @@ def _choose_provider(): def _choose_import_kb(): while True: c = input( - "Import knowledge base on startup? [Y/n]: " + "Import knowledge base on startup? Will improve results but takes about 30 minutes. [Y/n]: " ).strip().lower() if c in ("", "y", "yes"): From 02abd55df6a0266e835d365e3a8b32b897c8ab36 Mon Sep 17 00:00:00 2001 From: Vitaly Bogdanov Date: Thu, 11 Jun 2026 19:28:31 +0300 Subject: [PATCH 12/16] Remove embeddingprovider environment variable --- README.md | 2 +- entrypoint.sh | 4 ++-- scripts/import_knowledge.sh | 2 +- 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 7fca6c9e..20992a0b 100644 --- a/README.md +++ b/README.md @@ -86,7 +86,7 @@ After start go to https://webchat.quakenet.org/ to communicate with the agent. J If you are running OmegaClaw without Docker and would like to load it with preset knowledge, follow these steps: -1. Set embeddingprovider in your environment. It can be set to either OpenAI or Local. OpenAI embeddings also require OPENAI_API_KEY to be set in your environment. +1. Set EMBEDDING_PROVIDER in your environment. It can be set to either OpenAI or Local. OpenAI embeddings also require OPENAI_API_KEY to be set in your environment. 2. Run: ``` diff --git a/entrypoint.sh b/entrypoint.sh index dac745b3..64dee491 100644 --- a/entrypoint.sh +++ b/entrypoint.sh @@ -6,7 +6,7 @@ cd /PeTTa su www-data -s /bin/sh -c "sh /opt/nginx/nginx.sh" GATEWAY_URL="http://localhost:8080" -EMBEDDING_PROVIDER="${embeddingprovider:-Local}" +EMBEDDING_PROVIDER="${EMBEDDING_PROVIDER:-Local}" for arg in "$@"; do if [[ "$arg" == embeddingprovider=* ]]; then @@ -33,4 +33,4 @@ for var in $SAFE_VARS; do fi done -exec env -i $env_args su nobody -s /bin/sh -c "sh run.sh run.metta $*" \ No newline at end of file +exec env -i $env_args su nobody -s /bin/sh -c "sh run.sh run.metta $*" diff --git a/scripts/import_knowledge.sh b/scripts/import_knowledge.sh index 33cb0777..42af4a09 100644 --- a/scripts/import_knowledge.sh +++ b/scripts/import_knowledge.sh @@ -51,4 +51,4 @@ case "${PROVIDER}" in echo "Use embeddingprovider=OpenAI or embeddingprovider=Local." >&2 exit 1 ;; -esac \ No newline at end of file +esac From f157c379edc74cd68a222cc35f29a6bae6172416 Mon Sep 17 00:00:00 2001 From: Bereket-Eshete Date: Sat, 20 Jun 2026 14:24:45 +0300 Subject: [PATCH 13/16] refactor(channels): introduce BaseChannel and channels registry for DRY dispatch --- channels/base.py | 78 ++++ channels/channels_registry.py | 65 ++++ channels/irc.py | 287 +++++++------- channels/mattermost.py | 243 ++++-------- channels/slack.py | 696 ++++++++++++---------------------- channels/telegram.py | 391 ++++++++----------- 6 files changed, 755 insertions(+), 1005 deletions(-) create mode 100644 channels/base.py create mode 100644 channels/channels_registry.py diff --git a/channels/base.py b/channels/base.py new file mode 100644 index 00000000..7f327d99 --- /dev/null +++ b/channels/base.py @@ -0,0 +1,78 @@ +import abc +import os +import threading + + +class BaseChannel(abc.ABC): + def __init__(self): + self._last_message = "" + self._msg_lock = threading.Lock() + + self._auth_secret = "" + self._authenticated_id = None + self._auth_lock = threading.Lock() + + self._running = False + self._connected = False + self._thread = None + + def _set_last(self, msg: str) -> None: + with self._msg_lock: + if self._last_message == "": + self._last_message = msg + else: + self._last_message = self._last_message + " | " + msg + + def getLastMessage(self) -> str: + with self._msg_lock: + tmp = self._last_message + self._last_message = "" + return tmp + + def _set_auth_secret(self, secret=None) -> None: + if secret is None: + secret = os.environ.get("OMEGACLAW_AUTH_SECRET", "") + with self._auth_lock: + self._auth_secret = (secret or "").strip() + self._authenticated_id = None + + @staticmethod + def _parse_auth_candidate(msg: str) -> str: + text = msg.strip() + lower = text.lower() + if lower.startswith("auth "): + return text[5:].strip() + if lower.startswith("/auth "): + return text[6:].strip() + return text + + def _is_allowed_message(self, sender_id: str, msg: str) -> str: + candidate = self._parse_auth_candidate(msg) + with self._auth_lock: + if not self._auth_secret: + return "allow" + if candidate == self._auth_secret: + if self._authenticated_id is None: + self._authenticated_id = sender_id + return "auth_bound" + return "ignore" + if self._authenticated_id is None: + return "ignore" + return "allow" if sender_id == self._authenticated_id else "ignore" + + def start(self, auth_secret=None) -> threading.Thread: + self._set_auth_secret(auth_secret) + self._running = True + self._connected = False + self._thread = threading.Thread(target=self._run_loop, daemon=True) + self._thread.start() + return self._thread + + def stop(self) -> None: + self._running = False + + @abc.abstractmethod + def _run_loop(self) -> None: ... + + @abc.abstractmethod + def send_message(self, text: str) -> None: ... diff --git a/channels/channels_registry.py b/channels/channels_registry.py new file mode 100644 index 00000000..62a6e351 --- /dev/null +++ b/channels/channels_registry.py @@ -0,0 +1,65 @@ +import importlib + +_REGISTRY = { + "irc": ( + "channels.irc", "start_irc", + lambda token, channel_id, poll_interval, server_url: ( + channel_id or "##omegaclaw", + server_url or "irc.quakenet.org", + 6667, + "omegaclaw", + ), + ), + "telegram": ( + "channels.telegram", "start_telegram", + lambda token, channel_id, poll_interval, server_url: ( + token, + channel_id, + poll_interval, + ), + ), + "slack": ( + "channels.slack", "start_slack", + lambda token, channel_id, poll_interval, server_url: ( + token, + channel_id, + poll_interval, + ), + ), + "mattermost": ( + "channels.mattermost", "start_mattermost", + lambda token, channel_id, poll_interval, server_url: ( + server_url or "https://chat.singularitynet.io", + channel_id or "8fjrmabjx7gupy7e5kjznpt5qh", + token, + ), + ), + "mock": ( + "channels.mock", "start_mock", + lambda token, channel_id, poll_interval, server_url: (), + ), +} + +_active_module = None + + +def start(channel_name: str, token="", channel_id="", poll_interval=20, server_url=""): + global _active_module + channel_name = str(channel_name) + entry = _REGISTRY.get(channel_name) + if not entry: + raise ValueError(f"Unknown channel: {channel_name}") + module_path, start_fn, args_fn = entry + args = args_fn(str(token), str(channel_id), int(poll_interval), str(server_url)) + mod = importlib.import_module(module_path) + _active_module = mod + return getattr(mod, start_fn)(*args) + + +def getLastMessage() -> str: + return _active_module.getLastMessage() if _active_module else "" + + +def send_message(text: str) -> None: + if _active_module: + _active_module.send_message(text) diff --git a/channels/irc.py b/channels/irc.py index 89566db7..0497796d 100644 --- a/channels/irc.py +++ b/channels/irc.py @@ -1,163 +1,140 @@ -import os import random import socket +import textwrap import threading import time -import textwrap -import auth - -_running = False -_sock = None -_sock_lock = threading.Lock() -_last_message = "" -_msg_lock = threading.Lock() -_channel = None -_connected = False -_auth_lock = threading.Lock() -_authenticated_nick = None - -def _send(cmd): - with _sock_lock: - if _sock: - _sock.sendall((cmd + "\r\n").encode()) - time.sleep(1) - -def _set_last(msg): - global _last_message - with _msg_lock: - if _last_message == "": - _last_message = msg - else: - _last_message = _last_message + " | " + msg - -def getLastMessage(): - global _last_message - with _msg_lock: - tmp = _last_message - _last_message = "" - return tmp - - -def _normalize_nick(nick): - return nick.strip().lower() - - -def _parse_auth_candidate(msg): - text = msg.strip() - lower = text.lower() - if lower.startswith("auth "): - return text[5:].strip() - if lower.startswith("/auth "): - return text[6:].strip() - return text - -def _is_auth_command(msg): - lower = msg.strip().lower() - return lower.startswith("auth ") or lower.startswith("/auth ") - -def _is_allowed_message(nick, msg): - global _authenticated_nick - norm_nick = _normalize_nick(nick) - with _auth_lock: - if not auth.is_auth_enabled(): - return "allow" - if _authenticated_nick is not None: - return "allow" if norm_nick == _authenticated_nick else "ignore" - if not _is_auth_command(msg): - return "ignore" - candidate = _parse_auth_candidate(msg) - if auth.verify_token(candidate): - _authenticated_nick = norm_nick - return "auth_bound" - return "ignore" - -def _irc_loop(channel, server, port, nick): - global _running, _sock, _connected - print(f"[IRC] Connecting to {server}:{port} as {nick} for channel {channel}") - try: - sock = socket.create_connection((server, int(port)), timeout=15) - sock.settimeout(60) - print("[IRC] TCP connected") - except OSError as e: - print(f"[IRC] Connect failed: {e}") - return - _sock = sock - _send(f"NICK {nick}") - _send(f"USER {nick} 0 * :{nick}") - #_send(f"JOIN {channel}") - read_buffer = "" - while _running: + +from channels.base import BaseChannel + + +class IRCChannel(BaseChannel): + def __init__(self, channel, server, port, nick): + super().__init__() + if not channel.startswith("#"): + channel = f"#{channel}" + self._channel = channel + self._server = server + self._port = int(port) + self._nick = f"{nick}{random.randint(1000, 9999)}" + self._sock = None + self._sock_lock = threading.Lock() + + def _is_allowed_message(self, sender_id: str, msg: str) -> str: + norm_nick = sender_id.strip().lower() + candidate = self._parse_auth_candidate(msg) + with self._auth_lock: + if not self._auth_secret: + return "allow" + if candidate == self._auth_secret: + if self._authenticated_id is None: + self._authenticated_id = norm_nick + return "auth_bound" + return "ignore" + if self._authenticated_id is None: + return "ignore" + return "allow" if norm_nick == self._authenticated_id else "ignore" + + def _send_raw(self, cmd: str) -> None: + with self._sock_lock: + if self._sock: + self._sock.sendall((cmd + "\r\n").encode()) + time.sleep(1) + + def send_message(self, text: str) -> None: + segments = text.replace("\r", "").split("\\n") + lines = [] + for seg in segments: + lines.extend(textwrap.wrap(seg, width=400, break_long_words=True, break_on_hyphens=False)) + for chunk in lines: + try: + if self._connected and self._channel: + self._send_raw(f"PRIVMSG {self._channel} :{chunk}") + except Exception as e: + print(f"[IRC] send error: {e}") + + def _run_loop(self) -> None: + print(f"[IRC] Connecting to {self._server}:{self._port} as {self._nick}") try: - data = sock.recv(4096).decode(errors="ignore") - if not data: - break - except socket.timeout: - continue - except OSError: - break - read_buffer += data - while "\r\n" in read_buffer: - line, read_buffer = read_buffer.split("\r\n", 1) - if not line: + sock = socket.create_connection((self._server, self._port), timeout=15) + sock.settimeout(60) + except OSError as e: + print(f"[IRC] Connect failed: {e}") + return + + self._sock = sock + self._send_raw(f"NICK {self._nick}") + self._send_raw(f"USER {self._nick} 0 * :{self._nick}") + + buf = "" + while self._running: + try: + data = sock.recv(4096).decode(errors="ignore") + if not data: + break + except socket.timeout: continue - if line.startswith("PING"): - _send(f"PONG {line.split()[1]}") - parts = line.split() - if len(parts) > 1 and parts[1] == "001": - _connected = True - print(f"[IRC] Registered. Joining {_channel}") - _send(f"JOIN {_channel}") - elif len(parts) > 1 and parts[1] in {"403", "405", "471", "473", "474", "475"}: - print(f"[IRC] Join failed: {line}") - elif len(parts) > 1 and parts[1] == "433": - print(f"[IRC] Nickname in use: {line}") - elif line.startswith(":") and " PRIVMSG " in line: - try: - prefix, trailing = line[1:].split(" PRIVMSG ", 1) - nick = prefix.split("!", 1)[0] - - if " :" not in trailing: - continue # malformed, ignore safely - - msg = trailing.split(" :", 1)[1] - state = _is_allowed_message(nick, msg) - if state == "allow": - _set_last(f"{nick}: {msg}") - elif state == "auth_bound": - _send(f"PRIVMSG {_channel} :Authentication successful for {nick}.") - except Exception as e: - print(f"[IRC]: exception caught {repr(e)}") - _connected = False - with _sock_lock: - _sock = None - sock.close() - print("[IRC] Disconnected") - -def start_irc(channel, server="irc.libera.chat", port=6667, nick="omegaclaw"): - global _running, _channel, _connected - nick = f"{nick}{random.randint(1000, 9999)}" - if not channel.startswith("#"): - channel = f"#{channel}" - _running = True - _connected = False - _channel = channel - t = threading.Thread(target=_irc_loop, args=(channel, server, port, nick), daemon=True) - t.start() - return t + except OSError: + break + + buf += data + while "\r\n" in buf: + line, buf = buf.split("\r\n", 1) + if not line: + continue + if line.startswith("PING"): + self._send_raw(f"PONG {line.split()[1]}") + continue + parts = line.split() + if len(parts) < 2: + continue + if parts[1] == "001": + self._connected = True + print(f"[IRC] Registered. Joining {self._channel}") + self._send_raw(f"JOIN {self._channel}") + elif parts[1] in {"403", "405", "471", "473", "474", "475"}: + print(f"[IRC] Join failed: {line}") + elif parts[1] == "433": + print(f"[IRC] Nickname in use: {line}") + elif line.startswith(":") and " PRIVMSG " in line: + try: + prefix, trailing = line[1:].split(" PRIVMSG ", 1) + nick = prefix.split("!", 1)[0] + if " :" not in trailing: + continue # malformed, ignore safely + msg = trailing.split(" :", 1)[1] + state = self._is_allowed_message(nick, msg) + if state == "allow": + self._set_last(f"{nick}: {msg}") + elif state == "auth_bound": + self._send_raw(f"PRIVMSG {self._channel} :Authentication successful for {nick}.") + except Exception: + pass + + self._connected = False + with self._sock_lock: + self._sock = None + sock.close() + print("[IRC] Disconnected") + + +_instance: IRCChannel | None = None + + +def start_irc(channel, server="irc.quakenet.org", port=6667, nick="omegaclaw", auth_secret=None): + global _instance + _instance = IRCChannel(channel, server, port, nick) + return _instance.start(auth_secret) + def stop_irc(): - global _running - _running = False - -def send_message(text): - max_len = 400 - segments = text.replace("\r", "").split("\\n") - lines = [] - for segment in segments: - lines.extend(textwrap.wrap(segment, width=max_len, break_long_words=True, break_on_hyphens=False)) - for chunk in lines: - try: - if _connected and _channel: - _send(f"PRIVMSG {_channel} :{chunk}") - except Exception as e: - print(f"[IRC] error in send_message on channel {_channel}: {e}") + if _instance: + _instance.stop() + + +def getLastMessage() -> str: + return _instance.getLastMessage() if _instance else "" + + +def send_message(text: str) -> None: + if _instance: + _instance.send_message(text) diff --git a/channels/mattermost.py b/channels/mattermost.py index e13fe647..92370d0e 100644 --- a/channels/mattermost.py +++ b/channels/mattermost.py @@ -1,169 +1,92 @@ import json -import os -import threading import time - +import threading import requests import websocket -import auth - -_running = False -_ws = None -_ws_lock = threading.Lock() -_last_message = "" -_msg_lock = threading.Lock() -_connected = False -_auth_lock = threading.Lock() -_authenticated_user_id = None -_use_proxy = True - -# ---- Mattermost config (dummy token ok) ---- -MM_URL = "https://chat.singularitynet.io" -CHANNEL_ID = "8fjrmabjx7gupy7e5kjznpt5qh" #NOT AN ID JUST NAME: "omegaclaw"x -BOT_TOKEN = "" - -def _get_bot_user_id(): - global headers - r = requests.get( - f"{MM_URL}/api/v4/users/me", - headers=_headers - ) - return r.json()["id"] - -def _set_last(msg): - global _last_message - with _msg_lock: - if _last_message == "": - _last_message = msg - else: - _last_message = _last_message + " | " + msg - -def getLastMessage(): - global _last_message - with _msg_lock: - tmp = _last_message - _last_message = "" - return tmp - -def _parse_auth_candidate(msg): - text = msg.strip() - lower = text.lower() - if lower.startswith("auth "): - return text[5:].strip() - if lower.startswith("/auth "): - return text[6:].strip() - return text - -def _is_auth_command(msg): - lower = msg.strip().lower() - return lower.startswith("auth ") or lower.startswith("/auth ") - -def _is_allowed_message(user_id, msg): - global _authenticated_user_id - with _auth_lock: - if not auth.is_auth_enabled(): - return "allow" - if _authenticated_user_id is not None: - return "allow" if user_id == _authenticated_user_id else "ignore" - if not _is_auth_command(msg): - return "ignore" - candidate = _parse_auth_candidate(msg) - if auth.verify_token(candidate): - _authenticated_user_id = user_id - return "auth_bound" - return "ignore" - -def _get_display_name(user_id): - r = requests.get( - f"{MM_URL}/api/v4/users/{user_id}", - headers=_headers - ) - u = r.json() - # Mimic common Mattermost display setting - if u.get("first_name") or u.get("last_name"): - return f"{u.get('first_name','')} {u.get('last_name','')}".strip() +from channels.base import BaseChannel + +class MattermostChannel(BaseChannel): + def __init__(self, url, channel_id, bot_token): + super().__init__() + self._url = url.rstrip("/") + self._channel_id = channel_id + self._bot_token = bot_token + self._headers = {"Authorization": f"Bearer {bot_token}"} + self._bot_user_id = "" + + def _get(self, path): + return requests.get(f"{self._url}/api/v4{path}", headers=self._headers).json() + + def _get_bot_user_id(self): + return self._get("/users/me")["id"] + + def _get_display_name(self, user_id): + u = self._get(f"/users/{user_id}") + if u.get("first_name") or u.get("last_name"): + return f"{u.get('first_name','')} {u.get('last_name','')}".strip() + return u["username"] + + def _run_loop(self) -> None: + ws_url = self._url.replace("https", "wss") + "/api/v4/websocket" + ws = websocket.WebSocket() + ws.connect(ws_url, header=[f"Authorization: Bearer {self._bot_token}"]) + self._bot_user_id = self._get_bot_user_id() + self._connected = True + last_ping = time.time() + print("[MATTERMOST] Connected") + + while self._running: + try: + if time.time() - last_ping > 25: + ws.ping() + last_ping = time.time() + ws.settimeout(1) + event = json.loads(ws.recv()) + if event.get("event") == "posted": + post = json.loads(event["data"]["post"]) + if post["channel_id"] == self._channel_id \ + and post["user_id"] != self._bot_user_id: + user_id = post["user_id"] + msg = post.get("message", "") + state = self._is_allowed_message(user_id, msg) + if state == "allow": + name = self._get_display_name(user_id) + self._set_last(f"{name}: {msg}") + elif state == "auth_bound": + name = self._get_display_name(user_id) + self.send_message(f"Authentication successful for {name}.") + except websocket.WebSocketTimeoutException: + continue + except Exception: + break + + ws.close() + self._connected = False + print("[MATTERMOST] Disconnected") + + def send_message(self, text: str) -> None: + text = text.replace("\\n", "\n") + if not self._connected: + return + requests.post(f"{self._url}/api/v4/posts", headers=self._headers, + json={"channel_id": self._channel_id, "message": text}) + +_instance: MattermostChannel | None = None + +def start_mattermost(url, channel_id, bot_token, auth_secret=None): + global _instance + _instance = MattermostChannel(url, channel_id, bot_token) + return _instance.start(auth_secret) - return u["username"] - -def _ws_loop(): - global _ws, _connected, BOT_USER_ID, _use_proxy - - if _use_proxy: - ws_url = MM_URL.replace("http", "ws") - else: - ws_url = MM_URL.replace("https", "wss") - ws_url = ws_url + "/api/v4/websocket" - ws = websocket.WebSocket() - ws.connect(ws_url, header=[f"Authorization: Bearer {BOT_TOKEN}"]) - - BOT_USER_ID = _get_bot_user_id() - _ws = ws - _connected = True - - last_ping = time.time() - - while _running: - try: - # send ping every 25s - if time.time() - last_ping > 25: - ws.ping() - last_ping = time.time() - - ws.settimeout(1) - event = json.loads(ws.recv()) - - if event.get("event") == "posted": - post = json.loads(event["data"]["post"]) - if post["channel_id"] == CHANNEL_ID and post["user_id"] != BOT_USER_ID: - user_id = post["user_id"] - message = post.get("message", "") - state = _is_allowed_message(user_id, message) - if state == "allow": - name = _get_display_name(user_id) - _set_last(f"{name}: {message}") - elif state == "auth_bound": - name = _get_display_name(user_id) - send_message(f"Authentication successful for {name}.") - - except websocket.WebSocketTimeoutException: - continue - except Exception: - break - - ws.close() - _connected = False +def stop_mattermost(): + if _instance: + _instance.stop() -def start_mattermost(MM_URL_, CHANNEL_ID_): - global _running, MM_URL, CHANNEL_ID, BOT_TOKEN, _headers, _connected, _use_proxy - proxy = auth.get_proxy_url() - if proxy: - MM_URL = f"{proxy}/mattermost" - BOT_TOKEN = "proxy" - _headers = {} - _use_proxy = True - else: - MM_URL = MM_URL_ - BOT_TOKEN = os.environ.get("MM_BOT_TOKEN", "").strip() - _headers = {"Authorization": f"Bearer {BOT_TOKEN}"} if BOT_TOKEN else {} - _use_proxy = False - CHANNEL_ID = CHANNEL_ID_ - _running = True - _connected = False - t = threading.Thread(target=_ws_loop, daemon=True) - t.start() - return t +def getLastMessage() -> str: + return _instance.getLastMessage() if _instance else "" -def stop_mattermost(): - global _running - _running = False -def send_message(text): - text = text.replace("\\n", "\n") - if not _connected: - return - requests.post( - f"{MM_URL}/api/v4/posts", - headers=_headers, - json={"channel_id": CHANNEL_ID, "message": text} - ) +def send_message(text: str) -> None: + if _instance: + _instance.send_message(text) diff --git a/channels/slack.py b/channels/slack.py index e148641c..5c9340d2 100644 --- a/channels/slack.py +++ b/channels/slack.py @@ -1,474 +1,268 @@ import json -import os -import re import threading import time import urllib.error import urllib.parse import urllib.request -import auth -_running = False -_last_message = "" -_msg_lock = threading.Lock() -_state_lock = threading.Lock() +from channels.base import BaseChannel -_bot_token = "" -_channel_id = "" -_poll_interval = 10 -_bot_user_id = "" -_connected = False -_user_cache = {} -_channel_offsets = {} -_channel_name_cache = {} -_auto_bind_channels = [] -_auto_bind_index = 0 -_auto_bind_last_refresh = 0.0 - -_authenticated_user_id = None -_rate_limit_until = 0.0 _AUTO_BIND_REFRESH_INTERVAL = 300 -_SL_URL = "https://slack.com" - -class _SlackRateLimitError(Exception): +class _RateLimitError(Exception): def __init__(self, retry_after): - super().__init__(f"Slack rate limited (retry after {retry_after}s)") + super().__init__(f"Rate limited (retry after {retry_after}s)") self.retry_after = retry_after - -_URL_DISPLAY_RE = re.compile(r"<[^|>\s]+\|([^>]*)>") -_URL_BARE_RE = re.compile(r"<([^>\s]+)>") - - -def _slack_unwrap(text): - """Reverse Slack's outgoing auto-formatting so downstream layers see the original text.""" - if not text: - return text - text = _URL_DISPLAY_RE.sub(r"\1", text) - text = _URL_BARE_RE.sub(r"\1", text) - text = text.replace("<", "<").replace(">", ">").replace("&", "&") - return text - - -def _set_last(msg): - global _last_message - with _msg_lock: - if _last_message == "": - _last_message = msg - else: - _last_message = _last_message + " | " + msg - - -def getLastMessage(): - global _last_message - with _msg_lock: - tmp = _last_message - _last_message = "" - return tmp - - -def _parse_auth_candidate(msg): - text = msg.strip() - lower = text.lower() - if lower.startswith("auth "): - return text[5:].strip() - if lower.startswith("/auth "): - return text[6:].strip() - return text - -def _is_auth_command(msg): - lower = msg.strip().lower() - return lower.startswith("auth ") or lower.startswith("/auth ") - -def _channel_label(channel_id): - with _state_lock: - label = _channel_name_cache.get(channel_id) - return label or channel_id - - -def _is_allowed_message(channel_id, user_id, msg): - global _channel_id, _authenticated_user_id - with _state_lock: - if _channel_id and channel_id != _channel_id: - return "ignore" - if not auth.is_auth_enabled(): - if not _channel_id: - _bind_label = _channel_name_cache.get(channel_id, channel_id) - print(f"[SLACK] Auto-bound to channel {_bind_label}") - _channel_id = channel_id - return "allow" - if _authenticated_user_id is not None: - return "allow" if user_id == _authenticated_user_id else "ignore" - candidate = _parse_auth_candidate(msg) - if auth.verify_token(candidate): - _authenticated_user_id = user_id - _channel_id = channel_id - return "auth_bound" - return "ignore" - - -def _parse_retry_after(value): - try: - seconds = int(str(value).strip()) - return max(1, seconds) - except Exception: - return 60 - - -def _set_rate_limit_backoff(seconds): - global _rate_limit_until - until = time.time() + max(1, int(seconds)) - with _state_lock: - if until > _rate_limit_until: - _rate_limit_until = until - - -def _wait_for_rate_limit_window(): - with _state_lock: - wait = _rate_limit_until - time.time() - if wait > 0: - time.sleep(wait) - - -def _api_call(method, params=None, timeout=30): - global _SL_URL, _bot_token - - if not _bot_token: - raise RuntimeError("Slack adapter not initialized") - - params = params or {} - body = urllib.parse.urlencode(params).encode("utf-8") - req = urllib.request.Request( - f"{_SL_URL}/api/{method}", - data=body, - headers={ - "Authorization": f"Bearer {_bot_token}", - "Content-Type": "application/x-www-form-urlencoded", - }, - method="POST", - ) - - _wait_for_rate_limit_window() - try: - with urllib.request.urlopen(req, timeout=timeout) as response: - payload = json.loads(response.read().decode("utf-8", errors="ignore")) - response_headers = response.headers - except urllib.error.HTTPError as exc: - if exc.code == 429: - retry_after = _parse_retry_after(exc.headers.get("Retry-After")) - _set_rate_limit_backoff(retry_after) - raise _SlackRateLimitError(retry_after) from exc - raise - - if not payload.get("ok"): - err = payload.get("error", f"{method} failed") - if err == "ratelimited": - retry_after = _parse_retry_after(response_headers.get("Retry-After")) - _set_rate_limit_backoff(retry_after) - raise _SlackRateLimitError(retry_after) - raise RuntimeError(err) - - return payload - - -def _get_display_name(user_id): - with _state_lock: - cached = _user_cache.get(user_id) - if cached: - return cached - - name = user_id - try: - payload = _api_call("users.info", {"user": user_id}, timeout=15) - user = payload.get("user") or {} - profile = user.get("profile") or {} - - display_name = str(profile.get("display_name", "")).strip() - real_name = str(profile.get("real_name", "")).strip() - username = str(user.get("name", "")).strip() - - if display_name: - name = display_name - elif real_name: - name = real_name - elif username: - name = username - except Exception as exc: - print(f"[SLACK] Could not resolve user {user_id}: {exc}") - - with _state_lock: - _user_cache[user_id] = name - return name - - -def _cache_channel(channel): - channel_id = str(channel.get("id", "")).strip() - if not channel_id: - return - channel_name = str(channel.get("name", "")).strip() - label = f"#{channel_name}" if channel_name else channel_id - with _state_lock: - _channel_name_cache[channel_id] = label - - -def _initialize_identity(): - global _bot_user_id - payload = _api_call("auth.test", timeout=15) - bot_user_id = str(payload.get("user_id", "")).strip() - with _state_lock: - _bot_user_id = bot_user_id - - -def _validate_channel(): - payload = _api_call("conversations.info", {"channel": _channel_id}, timeout=15) - channel = payload.get("channel") or {} - _cache_channel(channel) - channel_name = str(channel.get("name", "")).strip() - if channel_name: - print(f"[SLACK] Channel ready: #{channel_name}") - else: - print(f"[SLACK] Channel ready: {_channel_id}") - - -def _list_joined_channels(): - channels = [] - cursor = "" - while True: - params = { - "types": "public_channel,private_channel", - "exclude_archived": "true", - "limit": 200, - } - if cursor: - params["cursor"] = cursor - - payload = _api_call("conversations.list", params=params, timeout=20) - for channel in payload.get("channels") or []: - if not channel.get("is_member"): - continue - channel_id = str(channel.get("id", "")).strip() - if not channel_id: - continue - _cache_channel(channel) - channels.append(channel_id) - - metadata = payload.get("response_metadata") or {} - cursor = str(metadata.get("next_cursor", "")).strip() - if not cursor: - break - return channels - - -def _initialize_cursor_for_channel(channel_id): - try: - payload = _api_call( - "conversations.history", - {"channel": channel_id, "limit": 1}, - timeout=15, +class SlackChannel(BaseChannel): + def __init__(self, bot_token, channel_id="", poll_interval=60): + super().__init__() + self._bot_token = str(bot_token).strip() + if not self._bot_token: + raise ValueError("SL_BOT_TOKEN is required") + self._channel_id = str(channel_id).strip() + self._poll_interval = max(1, int(poll_interval)) + self._bot_user_id = "" + self._user_cache: dict = {} + self._channel_offsets: dict = {} + self._channel_name_cache: dict = {} + self._auto_bind_channels: list = [] + self._auto_bind_index = 0 + self._auto_bind_last_refresh = 0.0 + self._rate_limit_until = 0.0 + + def _is_allowed_message(self, sender_id: str, msg: str, channel_id: str = "") -> str: + candidate = self._parse_auth_candidate(msg) + with self._auth_lock: + if self._channel_id and channel_id != self._channel_id: + return "ignore" + if not self._auth_secret: + if not self._channel_id: + label = self._channel_name_cache.get(channel_id, channel_id) + print(f"[SLACK] Auto-bound to channel {label}") + self._channel_id = channel_id + return "allow" + if candidate == self._auth_secret: + if self._authenticated_id is None: + self._authenticated_id = sender_id + self._channel_id = channel_id + return "auth_bound" + return "ignore" + if self._authenticated_id is None: + return "ignore" + return "allow" if sender_id == self._authenticated_id else "ignore" + + def _api_call(self, method, params=None, timeout=30): + params = params or {} + body = urllib.parse.urlencode(params).encode("utf-8") + req = urllib.request.Request( + f"https://slack.com/api/{method}", data=body, + headers={"Authorization": f"Bearer {self._bot_token}", + "Content-Type": "application/x-www-form-urlencoded"}, + method="POST", ) - messages = payload.get("messages") or [] - ts = "" - if messages: - ts = str(messages[0].get("ts", "")).strip() - with _state_lock: - _channel_offsets[channel_id] = ts - except Exception as exc: - print(f"[SLACK] Could not initialize cursor for {_channel_label(channel_id)}: {exc}") - - -def _initialize_auto_bind_cursors(): - channels = _refresh_auto_bind_channels(force=True) - if not channels: - print("[SLACK] Auto-bind waiting: no joined channels visible yet.") - else: - print(f"[SLACK] Auto-bind watching {len(channels)} joined channel(s).") - - -def _refresh_auto_bind_channels(force=False): - global _auto_bind_channels, _auto_bind_last_refresh, _auto_bind_index - now = time.time() - with _state_lock: - cached = list(_auto_bind_channels) - last_refresh = _auto_bind_last_refresh - - if (not force) and cached and (now - last_refresh) < _AUTO_BIND_REFRESH_INTERVAL: - return cached - - channels = _list_joined_channels() - with _state_lock: - _auto_bind_channels = channels - _auto_bind_last_refresh = now - if _auto_bind_index >= len(channels): - _auto_bind_index = 0 - return channels - - -def _next_auto_bind_channel(): - global _auto_bind_index - with _state_lock: - if not _auto_bind_channels: - return "" - channel_id = _auto_bind_channels[_auto_bind_index] - _auto_bind_index = (_auto_bind_index + 1) % len(_auto_bind_channels) - return channel_id - - -def _poll_channel(channel_id): - params = {"channel": channel_id, "limit": 15} - with _state_lock: - oldest = _channel_offsets.get(channel_id, "") - if oldest: - params["oldest"] = oldest - params["inclusive"] = "false" - - payload = _api_call("conversations.history", params=params, timeout=30) - messages = payload.get("messages") or [] - if not messages: - return - - ordered = sorted(messages, key=lambda m: float(m.get("ts", 0.0))) - max_ts = oldest - for message in ordered: - ts = str(message.get("ts", "")).strip() - if ts: - max_ts = ts - - # Ignore bot/system messages and process regular user text. - if message.get("subtype"): - continue - - text = _slack_unwrap(str(message.get("text", "")).strip()) - user_id = str(message.get("user", "")).strip() - if not text or not user_id: - continue - - with _state_lock: - bot_user_id = _bot_user_id - if bot_user_id and user_id == bot_user_id: - continue - - state = _is_allowed_message(channel_id, user_id, text) - display_name = _get_display_name(user_id) - if state == "allow": - _set_last(f"{display_name}: {text}") - elif state == "auth_bound": - send_message(f"Authentication successful for {display_name}.") - - if max_ts != oldest: - with _state_lock: - _channel_offsets[channel_id] = max_ts - - -def _poll_loop(): - global _connected - print("[SLACK] Polling started") - - while _running: + with self._auth_lock: + wait = self._rate_limit_until - time.time() + if wait > 0: + time.sleep(wait) try: - with _state_lock: - bound_channel = _channel_id - - if bound_channel: - with _state_lock: - known = bound_channel in _channel_offsets - if not known: - _initialize_cursor_for_channel(bound_channel) - _connected = True - else: - _poll_channel(bound_channel) - else: - channels = _refresh_auto_bind_channels() - if not channels: - channels = _refresh_auto_bind_channels(force=True) - channel_id = _next_auto_bind_channel() - if channel_id: - with _state_lock: - known = channel_id in _channel_offsets - if not known: - _initialize_cursor_for_channel(channel_id) - _connected = True - else: - _poll_channel(channel_id) - - _connected = True - except _SlackRateLimitError as exc: - _connected = False - print(f"[SLACK] Rate limited. Backing off for {exc.retry_after}s.") + with urllib.request.urlopen(req, timeout=timeout) as resp: + payload = json.loads(resp.read().decode("utf-8", errors="ignore")) + headers = resp.headers + except urllib.error.HTTPError as exc: + if exc.code == 429: + retry = max(1, int(exc.headers.get("Retry-After", 60))) + with self._auth_lock: + self._rate_limit_until = time.time() + retry + raise _RateLimitError(retry) from exc + raise + if not payload.get("ok"): + err = payload.get("error", f"{method} failed") + if err == "ratelimited": + retry = max(1, int(headers.get("Retry-After", 60))) + with self._auth_lock: + self._rate_limit_until = time.time() + retry + raise _RateLimitError(retry) + raise RuntimeError(err) + return payload + + def _get_display_name(self, user_id): + with self._auth_lock: + cached = self._user_cache.get(user_id) + if cached: + return cached + name = user_id + try: + payload = self._api_call("users.info", {"user": user_id}, timeout=15) + profile = (payload.get("user") or {}).get("profile") or {} + name = (profile.get("display_name") or profile.get("real_name") or + (payload["user"].get("name")) or user_id).strip() except Exception as exc: - _connected = False - print(f"[SLACK] Poll error: {exc}") - - time.sleep(max(1, int(_poll_interval))) - - _connected = False - print("[SLACK] Polling stopped") - - -def start_slack(channel_id, poll_interval=60): - global _running, _bot_token, _channel_id, _poll_interval, _connected - global _rate_limit_until, _auto_bind_channels, _auto_bind_index, _auto_bind_last_refresh - global _SL_URL - - proxy = auth.get_proxy_url() - if proxy: - _SL_URL = f"{proxy}/slack" - _bot_token = "proxy" - else: - _bot_token = os.environ.get("SL_BOT_TOKEN", "").strip() - if not _bot_token: - raise ValueError("SL_BOT_TOKEN is required") - - _channel_id = str(channel_id).strip() - - try: - _poll_interval = min(60, int(poll_interval)) - except Exception: - _poll_interval = 60 - - with _state_lock: - _user_cache.clear() - _channel_offsets.clear() - _channel_name_cache.clear() - _auto_bind_channels = [] - _auto_bind_index = 0 - _auto_bind_last_refresh = 0.0 - _rate_limit_until = 0.0 - _connected = False - _initialize_identity() - if _channel_id: - _validate_channel() - _initialize_cursor_for_channel(_channel_id) - else: - print("[SLACK] Starting adapter in auto-bind mode (channel not configured).") - _initialize_auto_bind_cursors() + print(f"[SLACK] Could not resolve user {user_id}: {exc}") + with self._auth_lock: + self._user_cache[user_id] = name + return name + + def _cache_channel(self, channel): + cid = str(channel.get("id", "")).strip() + name = str(channel.get("name", "")).strip() + if cid: + self._channel_name_cache[cid] = f"#{name}" if name else cid + + def _list_joined_channels(self): + channels, cursor = [], "" + while True: + params = {"types": "public_channel,private_channel", + "exclude_archived": "true", "limit": 200} + if cursor: + params["cursor"] = cursor + payload = self._api_call("conversations.list", params=params, timeout=20) + for ch in payload.get("channels") or []: + if ch.get("is_member"): + cid = str(ch.get("id", "")).strip() + if cid: + self._cache_channel(ch) + channels.append(cid) + cursor = str((payload.get("response_metadata") or {}).get("next_cursor", "")).strip() + if not cursor: + break + return channels + + def _init_cursor(self, channel_id): + try: + payload = self._api_call("conversations.history", + {"channel": channel_id, "limit": 1}, timeout=15) + msgs = payload.get("messages") or [] + ts = str(msgs[0].get("ts", "")).strip() if msgs else "" + self._channel_offsets[channel_id] = ts + except Exception as exc: + print(f"[SLACK] Could not initialize cursor for {channel_id}: {exc}") + + def _refresh_auto_bind(self, force=False): + now = time.time() + if (not force) and self._auto_bind_channels and \ + (now - self._auto_bind_last_refresh) < _AUTO_BIND_REFRESH_INTERVAL: + return self._auto_bind_channels + self._auto_bind_channels = self._list_joined_channels() + self._auto_bind_last_refresh = now + if self._auto_bind_index >= len(self._auto_bind_channels): + self._auto_bind_index = 0 + return self._auto_bind_channels + + def _next_auto_bind_channel(self): + if not self._auto_bind_channels: + return "" + cid = self._auto_bind_channels[self._auto_bind_index] + self._auto_bind_index = (self._auto_bind_index + 1) % len(self._auto_bind_channels) + return cid + + def _poll_channel(self, channel_id): + oldest = self._channel_offsets.get(channel_id, "") + params = {"channel": channel_id, "limit": 15} + if oldest: + params["oldest"] = oldest + params["inclusive"] = "false" + payload = self._api_call("conversations.history", params=params, timeout=30) + messages = sorted(payload.get("messages") or [], + key=lambda m: float(m.get("ts", 0.0))) + max_ts = oldest + for msg in messages: + ts = str(msg.get("ts", "")).strip() + if ts: + max_ts = ts + if msg.get("subtype"): + continue + text = str(msg.get("text", "")).strip() + user_id = str(msg.get("user", "")).strip() + if not text or not user_id or user_id == self._bot_user_id: + continue + state = self._is_allowed_message(user_id, text, channel_id) + name = self._get_display_name(user_id) + if state == "allow": + self._set_last(f"{name}: {text}") + elif state == "auth_bound": + self.send_message(f"Authentication successful for {name}.") + if max_ts != oldest: + self._channel_offsets[channel_id] = max_ts + + def _run_loop(self) -> None: + print("[SLACK] Polling started") + while self._running: + try: + with self._auth_lock: + bound = self._channel_id + if bound: + if bound not in self._channel_offsets: + self._init_cursor(bound) + else: + self._poll_channel(bound) + else: + channels = self._refresh_auto_bind() or self._refresh_auto_bind(force=True) + cid = self._next_auto_bind_channel() + if cid: + if cid not in self._channel_offsets: + self._init_cursor(cid) + else: + self._poll_channel(cid) + self._connected = True + except _RateLimitError as exc: + self._connected = False + print(f"[SLACK] Rate limited. Backing off {exc.retry_after}s.") + except Exception as exc: + self._connected = False + print(f"[SLACK] Poll error: {exc}") + time.sleep(max(1, self._poll_interval)) + self._connected = False + print("[SLACK] Polling stopped") + + def send_message(self, text: str) -> None: + text = str(text).replace("\\n", "\n").replace("\r", "") + if not text: + return + with self._auth_lock: + target = self._channel_id + if not target: + return + for i in range(0, len(text), 3900): + chunk = text[i:i + 3900] + try: + self._api_call("chat.postMessage", {"channel": target, "text": chunk}, timeout=15) + except Exception as exc: + print(f"[SLACK] Send failed: {exc}") + return + + def start(self, auth_secret=None) -> threading.Thread: + self._set_auth_secret(auth_secret) + payload = self._api_call("auth.test", timeout=15) + self._bot_user_id = str(payload.get("user_id", "")).strip() + if self._channel_id: + info = self._api_call("conversations.info", + {"channel": self._channel_id}, timeout=15) + self._cache_channel(info.get("channel") or {}) + self._init_cursor(self._channel_id) + print(f"[SLACK] Channel ready: " + f"{self._channel_name_cache.get(self._channel_id, self._channel_id)}") + else: + print("[SLACK] Starting in auto-bind mode.") + self._refresh_auto_bind(force=True) + print(f"[SLACK] Starting adapter with channel target: {self._channel_id or 'auto-bind'}") + return super().start(auth_secret=None) # secret already set above - _running = True - print(f"[SLACK] Starting adapter with channel target: {_channel_id or 'auto-bind'}") - t = threading.Thread(target=_poll_loop, daemon=True) - t.start() - return t +_instance: SlackChannel | None = None +def start_slack(bot_token, channel_id="", poll_interval=60, auth_secret=None): + global _instance + _instance = SlackChannel(bot_token, channel_id, poll_interval) + return _instance.start(auth_secret) def stop_slack(): - global _running - _running = False + if _instance: + _instance.stop() +def getLastMessage() -> str: + return _instance.getLastMessage() if _instance else "" -def send_message(text): - text = str(text).replace("\\n", "\n").replace("\r", "") - if not text: - return - if not _channel_id: - return - max_len = 3900 - for i in range(0, len(text), max_len): - chunk = text[i:i + max_len] - if not chunk: - continue - try: - _api_call("chat.postMessage", {"channel": _channel_id, "text": chunk}, timeout=15) - except Exception as exc: - print(f"[SLACK] Send failed: {exc}") - return +def send_message(text: str) -> None: + if _instance: + _instance.send_message(text) diff --git a/channels/telegram.py b/channels/telegram.py index a91f8326..dbb7f020 100644 --- a/channels/telegram.py +++ b/channels/telegram.py @@ -1,253 +1,166 @@ import json -import os import threading import time import urllib.parse import urllib.request -import auth -_running = False -_last_message = "" -_msg_lock = threading.Lock() -_state_lock = threading.Lock() +from channels.base import BaseChannel -_bot_token = "" -_api_base = "" -_chat_id = "" -_poll_timeout = 20 -_offset = None -_connected = False - -_authenticated_user_id = None - - -def _set_last(msg): - global _last_message - with _msg_lock: - if _last_message == "": - _last_message = msg - else: - _last_message = _last_message + " | " + msg - - -def getLastMessage(): - global _last_message - with _msg_lock: - tmp = _last_message - _last_message = "" - return tmp - - -def _parse_auth_candidate(msg): - text = msg.strip() - lower = text.lower() - if lower.startswith("auth "): - return text[5:].strip() - if lower.startswith("/auth "): - return text[6:].strip() - return text - - -def _display_name(user, chat): - username = str(user.get("username", "")).strip() - if username: - return f"@{username}" - - first = str(user.get("first_name", "")).strip() - last = str(user.get("last_name", "")).strip() - full = f"{first} {last}".strip() - if full: - return full - - title = str(chat.get("title", "")).strip() - if title: - return title - - return "telegram_user" - - -def _api_call(method, params=None, timeout=30, use_post=False): - if not _api_base: - raise RuntimeError("Telegram adapter not initialized") - - params = params or {} - encoded = urllib.parse.urlencode(params).encode("utf-8") - url = f"{_api_base}/{method}" - - if use_post: - req = urllib.request.Request(url, data=encoded) - else: - if params: - url = f"{url}?{urllib.parse.urlencode(params)}" - req = urllib.request.Request(url) - - with urllib.request.urlopen(req, timeout=timeout) as response: - payload = json.loads(response.read().decode("utf-8", errors="ignore")) - - if not payload.get("ok"): - raise RuntimeError(payload.get("description", f"{method} failed")) - - return payload.get("result") - - -def _initialize_offset(): - global _offset - try: - updates = _api_call("getUpdates", {"timeout": 0}, timeout=10) or [] - except Exception as exc: - print(f"[TELEGRAM] Could not read initial offset: {exc}") - return - - max_update = -1 - for update in updates: - update_id = update.get("update_id") - if isinstance(update_id, int): - max_update = max(max_update, update_id) - - if max_update >= 0: - with _state_lock: - _offset = max_update + 1 - - -def _is_auth_command(msg): - lower = msg.strip().lower() - return lower.startswith("auth ") or lower.startswith("/auth ") - - -def _is_allowed_message(chat_id, user_id, msg): - global _chat_id, _authenticated_user_id - - with _state_lock: - if _chat_id and chat_id != _chat_id: - return "ignore" - if not auth.is_auth_enabled(): - if not _chat_id: - _chat_id = chat_id - return "allow" - if _authenticated_user_id is not None: - if chat_id != _chat_id: +class TelegramChannel(BaseChannel): + def __init__(self, bot_token, chat_id="", poll_timeout=20): + super().__init__() + self._bot_token = str(bot_token).strip() + if not self._bot_token: + raise ValueError("TG_BOT_TOKEN is required") + self._api_base = f"https://api.telegram.org/bot{self._bot_token}" + self._chat_id = str(chat_id).strip() + self._poll_timeout = max(1, int(poll_timeout)) + self._offset = None + self._state_lock = threading.Lock() + self._authenticated_chat_id = None + + def _set_auth_secret(self, secret=None) -> None: + super()._set_auth_secret(secret) + with self._auth_lock: + self._authenticated_chat_id = None + + def _is_allowed_message(self, sender_id: str, msg: str, chat_id: str = "") -> str: + candidate = self._parse_auth_candidate(msg) + with self._auth_lock: + if self._chat_id and chat_id != self._chat_id: return "ignore" - return "allow" if user_id == _authenticated_user_id else "ignore" - if not _is_auth_command(msg): - return "ignore" - candidate = _parse_auth_candidate(msg) - if auth.verify_token(candidate): - _authenticated_user_id = user_id - _chat_id = chat_id - return "auth_bound" - return "ignore" - - -def _poll_loop(): - global _connected, _offset - print("[TELEGRAM] Polling started") - - while _running: + if not self._auth_secret: + if not self._chat_id: + self._chat_id = chat_id + return "allow" + if self._authenticated_id is None: + if candidate == self._auth_secret: + self._authenticated_id = sender_id + self._authenticated_chat_id = chat_id + self._chat_id = chat_id + return "auth_bound" + return "ignore" + if chat_id != self._authenticated_chat_id: + return "ignore" + return "allow" if sender_id == self._authenticated_id else "ignore" + + def _api_call(self, method, params=None, timeout=30, use_post=False): + params = params or {} + url = f"{self._api_base}/{method}" + encoded = urllib.parse.urlencode(params).encode("utf-8") + if use_post: + req = urllib.request.Request(url, data=encoded) + else: + if params: + url = f"{url}?{urllib.parse.urlencode(params)}" + req = urllib.request.Request(url) + with urllib.request.urlopen(req, timeout=timeout) as resp: + payload = json.loads(resp.read().decode("utf-8", errors="ignore")) + if not payload.get("ok"): + raise RuntimeError(payload.get("description", f"{method} failed")) + return payload.get("result") + + def _initialize_offset(self): try: - params = {"timeout": int(_poll_timeout)} - with _state_lock: - if _offset is not None: - params["offset"] = _offset - - updates = _api_call("getUpdates", params=params, timeout=int(_poll_timeout) + 10) or [] - _connected = True - - for update in updates: - update_id = update.get("update_id") - if isinstance(update_id, int): - with _state_lock: - if _offset is None or (update_id + 1) > _offset: - _offset = update_id + 1 - - message = update.get("message") or update.get("edited_message") - if not isinstance(message, dict): - continue - - text = message.get("text") - if not text: - continue - - chat = message.get("chat") or {} - user = message.get("from") or {} - chat_id = str(chat.get("id", "")).strip() - user_id = str(user.get("id", "")).strip() - if not chat_id or not user_id: - continue - - state = _is_allowed_message(chat_id, user_id, text) - display_name = _display_name(user, chat) - if state == "allow": - _set_last(f"{display_name}: {text}") - elif state == "auth_bound": - send_message(f"Authentication successful for {display_name}.") + updates = self._api_call("getUpdates", {"timeout": 0}, timeout=10) or [] except Exception as exc: - _connected = False - print(f"[TELEGRAM] Poll error: {exc}") - time.sleep(2) - - _connected = False - print("[TELEGRAM] Polling stopped") - - -def start_telegram(chat_id="", poll_timeout=20): - global _running, _bot_token, _api_base, _chat_id, _poll_timeout, _offset, _connected - - proxy = auth.get_proxy_url() - if proxy: - _bot_token = "proxy" - _api_base = f"{proxy}/telegram" - else: - _bot_token = os.environ.get("TG_BOT_TOKEN", "").strip() - if not _bot_token: - raise ValueError("TG_BOT_TOKEN is required") - _api_base = f"https://api.telegram.org/bot{_bot_token}" - - _chat_id = str(chat_id).strip() - - try: - _poll_timeout = max(1, int(poll_timeout)) - except Exception: - _poll_timeout = 20 - - _offset = None - _running = True - _connected = False - print(f"[TELEGRAM] Starting adapter with chat target: {_chat_id or 'auto-bind'}") - _initialize_offset() - - t = threading.Thread(target=_poll_loop, daemon=True) - t.start() - return t - + print(f"[TELEGRAM] Could not read initial offset: {exc}") + return + max_update = max((u.get("update_id", -1) for u in updates), default=-1) + if max_update >= 0: + with self._state_lock: + self._offset = max_update + 1 + + @staticmethod + def _display_name(user, chat): + username = str(user.get("username", "")).strip() + if username: + return f"@{username}" + full = f"{user.get('first_name','')} {user.get('last_name','')}".strip() + if full: + return full + title = str(chat.get("title", "")).strip() + return title or "telegram_user" + + def _run_loop(self) -> None: + print("[TELEGRAM] Polling started") + while self._running: + try: + params = {"timeout": self._poll_timeout} + with self._state_lock: + if self._offset is not None: + params["offset"] = self._offset + updates = self._api_call("getUpdates", params=params, + timeout=self._poll_timeout + 10) or [] + self._connected = True + for update in updates: + uid = update.get("update_id") + if isinstance(uid, int): + with self._state_lock: + if self._offset is None or uid + 1 > self._offset: + self._offset = uid + 1 + message = update.get("message") or update.get("edited_message") + if not isinstance(message, dict): + continue + text = message.get("text") + if not text: + continue + chat = message.get("chat") or {} + user = message.get("from") or {} + chat_id = str(chat.get("id", "")).strip() + user_id = str(user.get("id", "")).strip() + if not chat_id or not user_id: + continue + state = self._is_allowed_message(user_id, text, chat_id) + name = self._display_name(user, chat) + if state == "allow": + self._set_last(f"{name}: {text}") + elif state == "auth_bound": + self.send_message(f"Authentication successful for {name}.") + except Exception as exc: + self._connected = False + print(f"[TELEGRAM] Poll error: {exc}") + time.sleep(2) + self._connected = False + print("[TELEGRAM] Polling stopped") + + def send_message(self, text: str) -> None: + text = str(text).replace("\\n", "\n").replace("\r", "") + if not text: + return + with self._auth_lock: + target = self._chat_id + if not self._connected or not target: + return + for i in range(0, len(text), 3900): + chunk = text[i:i + 3900] + try: + self._api_call("sendMessage", {"chat_id": target, "text": chunk}, + timeout=15, use_post=True) + except Exception as exc: + print(f"[TELEGRAM] Send failed: {exc}") + return + + def start(self, auth_secret=None) -> threading.Thread: + self._set_auth_secret(auth_secret) + print(f"[TELEGRAM] Starting adapter with chat target: {self._chat_id or 'auto-bind'}") + self._initialize_offset() + return super().start(auth_secret=None) # secret already set above + +_instance: TelegramChannel | None = None + +def start_telegram(bot_token, chat_id="", poll_timeout=20, auth_secret=None): + global _instance + _instance = TelegramChannel(bot_token, chat_id, poll_timeout) + return _instance.start(auth_secret) def stop_telegram(): - global _running - _running = False - - -def send_message(text): - text = str(text).replace("\\n", "\n").replace("\r", "") - if not text: - return - - with _state_lock: - target_chat = _chat_id + if _instance: + _instance.stop() - if not _connected or not target_chat: - return +def getLastMessage() -> str: + return _instance.getLastMessage() if _instance else "" - max_len = 3900 - for i in range(0, len(text), max_len): - chunk = text[i:i + max_len] - if not chunk: - continue - try: - _api_call( - "sendMessage", - {"chat_id": target_chat, "text": chunk}, - timeout=15, - use_post=True, - ) - except Exception as exc: - print(f"[TELEGRAM] Send failed: {exc}") - return +def send_message(text: str) -> None: + if _instance: + _instance.send_message(text) From 80a6a30dd3a17c5fc5d1610d4a47d58d1406fcdb Mon Sep 17 00:00:00 2001 From: Bereket-Eshete Date: Sat, 20 Jun 2026 14:25:18 +0300 Subject: [PATCH 14/16] refactor(channels.metta): simplify channel config and dispatch via channels registry --- lib_omegaclaw.metta | 2 ++ src/channels.metta | 78 ++++++++++++++------------------------------- 2 files changed, 26 insertions(+), 54 deletions(-) diff --git a/lib_omegaclaw.metta b/lib_omegaclaw.metta index 7c5ca790..76341d15 100644 --- a/lib_omegaclaw.metta +++ b/lib_omegaclaw.metta @@ -8,11 +8,13 @@ !(import! &self (library OmegaClaw-Core ./src/utils)) !(import! &self (library OmegaClaw-Core ./src/helper.py)) !(import! &self (library OmegaClaw-Core ./src/agentverse.py)) +!(import! &self (library OmegaClaw-Core ./channels/base.py)) !(import! &self (library OmegaClaw-Core ./channels/irc.py)) !(import! &self (library OmegaClaw-Core ./channels/mattermost.py)) !(import! &self (library OmegaClaw-Core ./channels/telegram.py)) !(import! &self (library OmegaClaw-Core ./channels/slack.py)) !(import! &self (library OmegaClaw-Core ./channels/websearch.py)) +!(import! &self (library OmegaClaw-Core ./channels/channels_registry.py)) !(import! &self (library OmegaClaw-Core ./src/channels)) !(import! &self (library OmegaClaw-Core ./src/skills)) !(import! &self (library OmegaClaw-Core ./src/memory)) diff --git a/src/channels.metta b/src/channels.metta index 95e642e4..d99a6d41 100644 --- a/src/channels.metta +++ b/src/channels.metta @@ -1,68 +1,38 @@ ;configured at runtime: -(= (IRC_channel) (empty)) -(= (IRC_server) (empty)) -(= (IRC_port) (empty)) -(= (IRC_user) (empty)) -(= (commchannel) (empty)) -(= (MM_URL) (empty)) -(= (MM_CHANNEL_ID) (empty)) -(= (TG_CHAT_ID) (empty)) -(= (TG_POLL_TIMEOUT) (empty)) -(= (SL_CHANNEL_ID) (empty)) -(= (SL_POLL_INTERVAL) (empty)) +(= (commchannel) (empty)) +(= (bot_token) (empty)) +(= (channel_id) (empty)) +(= (poll_interval) (empty)) +(= (server_url) (empty)) -;Connect all the channels: +;Connect the configured channel — MeTTa resolves config, Python starts it: (= (initChannels) (progn (println! "Initializing channels") - (configure commchannel irc) - (if (== (commchannel) irc) - (progn (configure IRC_channel ##omegaclaw) - (configure IRC_server "irc.quakenet.org") - (configure IRC_port 6667) - (configure IRC_user omegaclaw) - (py-call (irc.start_irc (IRC_channel) (IRC_server) (IRC_port) (IRC_user)))) - (if (== (commchannel) telegram) - (progn (configure TG_CHAT_ID "") - (configure TG_POLL_TIMEOUT 20) - (py-call (telegram.start_telegram (TG_CHAT_ID) (TG_POLL_TIMEOUT)))) - (if (== (commchannel) slack) - (progn (configure SL_CHANNEL_ID "") - (configure SL_POLL_INTERVAL 60) - (py-call (slack.start_slack (SL_CHANNEL_ID) (SL_POLL_INTERVAL)))) - (if (== (commchannel) mattermost) - (progn (configure MM_URL "https://chat.singularitynet.io") - (configure MM_CHANNEL_ID "8fjrmabjx7gupy7e5kjznpt5qh") - (py-call (mattermost.start_mattermost (MM_URL) (MM_CHANNEL_ID)))) - (py-call (mock.start_mock)))))))) + (configure commchannel irc) + (configure bot_token "") + (configure channel_id "") ; channel_id and chat_id can be used interchangably, depending on the channel type + (configure poll_interval 20) + (configure server_url "") + (py-call (channels_registry.start + (commchannel) + (bot_token) + (channel_id) + (poll_interval) + (server_url))))) -;Receive the latest user message considering all communication channels: +;Receive the latest user message: (= (receive) - (if (== (commchannel) irc) - (py-call (irc.getLastMessage)) - (if (== (commchannel) telegram) - (py-call (telegram.getLastMessage)) - (if (== (commchannel) slack) - (py-call (slack.getLastMessage)) - (if (== (commchannel) mattermost) - (py-call (mattermost.getLastMessage)) - (py-call (mock.getLastMessage))))))) + (py-call (channels_registry.getLastMessage))) -;Send a message to all communication channels: +;Send a message: !(change-state! &lastsend "") (= (send $msg) (if (!= $msg (get-state &lastsend)) (progn (change-state! &lastsend $msg) - (let $safemsg (string-replace $msg "\n" "\\n") - (if (== (commchannel) irc) - (let $temp (cut) (py-call (irc.send_message $safemsg))) - (if (== (commchannel) telegram) - (let $temp (cut) (py-call (telegram.send_message $safemsg))) - (if (== (commchannel) slack) - (let $temp (cut) (py-call (slack.send_message $safemsg))) - (if (== (commchannel) mattermost) - (let $temp (cut) (py-call (mattermost.send_message $safemsg))) - (let $temp (cut) (py-call (mock.send_message $safemsg))))))))) _)) + (let $safemsg (string-replace $msg "\n" "\\n") + (py-call (channels_registry.send_message $safemsg)))) + _)) ;Search the internet for some information: (= (search $msg) - (py-call (websearch.search $msg))) + (py-call (websearch.search $msg))) \ No newline at end of file From 8b86b9b0c6846b2e798b50e6076b50e125ec3775 Mon Sep 17 00:00:00 2001 From: Bereket-Eshete Date: Wed, 24 Jun 2026 12:04:13 +0300 Subject: [PATCH 15/16] fix(channels): preserve proxy-based auth after channel refactor Restore auth.is_auth_enabled() + auth.verify_token() proxy logic in BaseChannel._is_allowed_message, replacing the OMEGACLAW_AUTH_SECRET env-var approach introduced by the refactor. Restore proxy API base URL overrides in start() for Telegram ({proxy}/telegram), Slack ({proxy}/slack), and Mattermost ({proxy}/mattermost) to match pre-refactor behaviour. --- channels/base.py | 32 ++++++++------------- channels/channels_registry.py | 7 +++-- channels/irc.py | 23 ++++++++-------- channels/mattermost.py | 21 ++++++++++---- channels/slack.py | 52 +++++++++++++++++++++++------------ channels/telegram.py | 43 ++++++++++++++--------------- 6 files changed, 97 insertions(+), 81 deletions(-) diff --git a/channels/base.py b/channels/base.py index 7f327d99..fc8df576 100644 --- a/channels/base.py +++ b/channels/base.py @@ -1,14 +1,14 @@ import abc -import os import threading +import auth + class BaseChannel(abc.ABC): def __init__(self): self._last_message = "" self._msg_lock = threading.Lock() - self._auth_secret = "" self._authenticated_id = None self._auth_lock = threading.Lock() @@ -29,13 +29,6 @@ def getLastMessage(self) -> str: self._last_message = "" return tmp - def _set_auth_secret(self, secret=None) -> None: - if secret is None: - secret = os.environ.get("OMEGACLAW_AUTH_SECRET", "") - with self._auth_lock: - self._auth_secret = (secret or "").strip() - self._authenticated_id = None - @staticmethod def _parse_auth_candidate(msg: str) -> str: text = msg.strip() @@ -47,21 +40,18 @@ def _parse_auth_candidate(msg: str) -> str: return text def _is_allowed_message(self, sender_id: str, msg: str) -> str: - candidate = self._parse_auth_candidate(msg) with self._auth_lock: - if not self._auth_secret: + if not auth.is_auth_enabled(): return "allow" - if candidate == self._auth_secret: - if self._authenticated_id is None: - self._authenticated_id = sender_id - return "auth_bound" - return "ignore" - if self._authenticated_id is None: - return "ignore" - return "allow" if sender_id == self._authenticated_id else "ignore" + if self._authenticated_id is not None: + return "allow" if sender_id == self._authenticated_id else "ignore" + candidate = self._parse_auth_candidate(msg) + if auth.verify_token(candidate): + self._authenticated_id = sender_id + return "auth_bound" + return "ignore" - def start(self, auth_secret=None) -> threading.Thread: - self._set_auth_secret(auth_secret) + def start(self) -> threading.Thread: self._running = True self._connected = False self._thread = threading.Thread(target=self._run_loop, daemon=True) diff --git a/channels/channels_registry.py b/channels/channels_registry.py index 62a6e351..1d3474bc 100644 --- a/channels/channels_registry.py +++ b/channels/channels_registry.py @@ -1,4 +1,5 @@ import importlib +import os _REGISTRY = { "irc": ( @@ -13,7 +14,7 @@ "telegram": ( "channels.telegram", "start_telegram", lambda token, channel_id, poll_interval, server_url: ( - token, + token or os.environ.get("TG_BOT_TOKEN", ""), channel_id, poll_interval, ), @@ -21,7 +22,7 @@ "slack": ( "channels.slack", "start_slack", lambda token, channel_id, poll_interval, server_url: ( - token, + token or os.environ.get("SL_BOT_TOKEN", ""), channel_id, poll_interval, ), @@ -31,7 +32,7 @@ lambda token, channel_id, poll_interval, server_url: ( server_url or "https://chat.singularitynet.io", channel_id or "8fjrmabjx7gupy7e5kjznpt5qh", - token, + token or os.environ.get("MM_BOT_TOKEN", ""), ), ), "mock": ( diff --git a/channels/irc.py b/channels/irc.py index 0497796d..8b2a8c43 100644 --- a/channels/irc.py +++ b/channels/irc.py @@ -4,6 +4,7 @@ import threading import time +import auth from channels.base import BaseChannel @@ -21,18 +22,16 @@ def __init__(self, channel, server, port, nick): def _is_allowed_message(self, sender_id: str, msg: str) -> str: norm_nick = sender_id.strip().lower() - candidate = self._parse_auth_candidate(msg) with self._auth_lock: - if not self._auth_secret: + if not auth.is_auth_enabled(): return "allow" - if candidate == self._auth_secret: - if self._authenticated_id is None: - self._authenticated_id = norm_nick - return "auth_bound" - return "ignore" - if self._authenticated_id is None: - return "ignore" - return "allow" if norm_nick == self._authenticated_id else "ignore" + if self._authenticated_id is not None: + return "allow" if norm_nick == self._authenticated_id else "ignore" + candidate = self._parse_auth_candidate(msg) + if auth.verify_token(candidate): + self._authenticated_id = norm_nick + return "auth_bound" + return "ignore" def _send_raw(self, cmd: str) -> None: with self._sock_lock: @@ -120,10 +119,10 @@ def _run_loop(self) -> None: _instance: IRCChannel | None = None -def start_irc(channel, server="irc.quakenet.org", port=6667, nick="omegaclaw", auth_secret=None): +def start_irc(channel, server="irc.quakenet.org", port=6667, nick="omegaclaw"): global _instance _instance = IRCChannel(channel, server, port, nick) - return _instance.start(auth_secret) + return _instance.start() def stop_irc(): diff --git a/channels/mattermost.py b/channels/mattermost.py index 92370d0e..76fa4e67 100644 --- a/channels/mattermost.py +++ b/channels/mattermost.py @@ -4,6 +4,7 @@ import requests import websocket +import auth from channels.base import BaseChannel class MattermostChannel(BaseChannel): @@ -28,9 +29,12 @@ def _get_display_name(self, user_id): return u["username"] def _run_loop(self) -> None: - ws_url = self._url.replace("https", "wss") + "/api/v4/websocket" + ws_url = (self._url.replace("http", "ws") if self._bot_token == "proxy" + else self._url.replace("https", "wss")) + "/api/v4/websocket" ws = websocket.WebSocket() - ws.connect(ws_url, header=[f"Authorization: Bearer {self._bot_token}"]) + ws_headers = ([f"Authorization: Bearer {self._bot_token}"] + if self._bot_token and self._bot_token != "proxy" else []) + ws.connect(ws_url, header=ws_headers) self._bot_user_id = self._get_bot_user_id() self._connected = True last_ping = time.time() @@ -74,10 +78,17 @@ def send_message(self, text: str) -> None: _instance: MattermostChannel | None = None -def start_mattermost(url, channel_id, bot_token, auth_secret=None): +def start_mattermost(url, channel_id, bot_token): global _instance - _instance = MattermostChannel(url, channel_id, bot_token) - return _instance.start(auth_secret) + proxy = auth.get_proxy_url() + if proxy: + url = f"{proxy}/mattermost" + bot_token = "proxy" + _instance = MattermostChannel(url, channel_id, bot_token) + _instance._headers = {} + else: + _instance = MattermostChannel(url, channel_id, bot_token) + return _instance.start() def stop_mattermost(): if _instance: diff --git a/channels/slack.py b/channels/slack.py index 5c9340d2..09e6285f 100644 --- a/channels/slack.py +++ b/channels/slack.py @@ -1,14 +1,28 @@ import json +import re import threading import time import urllib.error import urllib.parse import urllib.request +import auth from channels.base import BaseChannel _AUTO_BIND_REFRESH_INTERVAL = 300 +_URL_DISPLAY_RE = re.compile(r"<[^|>\s]+\|([^>]*)>") +_URL_BARE_RE = re.compile(r"<([^>\s]+)>") + + +def _slack_unwrap(text): + if not text: + return text + text = _URL_DISPLAY_RE.sub(r"\1", text) + text = _URL_BARE_RE.sub(r"\1", text) + text = text.replace("<", "<").replace(">", ">").replace("&", "&") + return text + class _RateLimitError(Exception): def __init__(self, retry_after): super().__init__(f"Rate limited (retry after {retry_after}s)") @@ -20,6 +34,7 @@ def __init__(self, bot_token, channel_id="", poll_interval=60): self._bot_token = str(bot_token).strip() if not self._bot_token: raise ValueError("SL_BOT_TOKEN is required") + self._api_base = "https://slack.com" self._channel_id = str(channel_id).strip() self._poll_interval = max(1, int(poll_interval)) self._bot_user_id = "" @@ -32,31 +47,29 @@ def __init__(self, bot_token, channel_id="", poll_interval=60): self._rate_limit_until = 0.0 def _is_allowed_message(self, sender_id: str, msg: str, channel_id: str = "") -> str: - candidate = self._parse_auth_candidate(msg) with self._auth_lock: if self._channel_id and channel_id != self._channel_id: return "ignore" - if not self._auth_secret: + if not auth.is_auth_enabled(): if not self._channel_id: label = self._channel_name_cache.get(channel_id, channel_id) print(f"[SLACK] Auto-bound to channel {label}") self._channel_id = channel_id return "allow" - if candidate == self._auth_secret: - if self._authenticated_id is None: - self._authenticated_id = sender_id - self._channel_id = channel_id - return "auth_bound" - return "ignore" - if self._authenticated_id is None: - return "ignore" - return "allow" if sender_id == self._authenticated_id else "ignore" + if self._authenticated_id is not None: + return "allow" if sender_id == self._authenticated_id else "ignore" + candidate = self._parse_auth_candidate(msg) + if auth.verify_token(candidate): + self._authenticated_id = sender_id + self._channel_id = channel_id + return "auth_bound" + return "ignore" def _api_call(self, method, params=None, timeout=30): params = params or {} body = urllib.parse.urlencode(params).encode("utf-8") req = urllib.request.Request( - f"https://slack.com/api/{method}", data=body, + f"{self._api_base}/api/{method}", data=body, headers={"Authorization": f"Bearer {self._bot_token}", "Content-Type": "application/x-www-form-urlencoded"}, method="POST", @@ -172,7 +185,7 @@ def _poll_channel(self, channel_id): max_ts = ts if msg.get("subtype"): continue - text = str(msg.get("text", "")).strip() + text = _slack_unwrap(str(msg.get("text", "")).strip()) user_id = str(msg.get("user", "")).strip() if not text or not user_id or user_id == self._bot_user_id: continue @@ -231,8 +244,11 @@ def send_message(self, text: str) -> None: print(f"[SLACK] Send failed: {exc}") return - def start(self, auth_secret=None) -> threading.Thread: - self._set_auth_secret(auth_secret) + def start(self) -> threading.Thread: + proxy = auth.get_proxy_url() + if proxy: + self._bot_token = "proxy" + self._api_base = f"{proxy}/slack" payload = self._api_call("auth.test", timeout=15) self._bot_user_id = str(payload.get("user_id", "")).strip() if self._channel_id: @@ -246,14 +262,14 @@ def start(self, auth_secret=None) -> threading.Thread: print("[SLACK] Starting in auto-bind mode.") self._refresh_auto_bind(force=True) print(f"[SLACK] Starting adapter with channel target: {self._channel_id or 'auto-bind'}") - return super().start(auth_secret=None) # secret already set above + return super().start() _instance: SlackChannel | None = None -def start_slack(bot_token, channel_id="", poll_interval=60, auth_secret=None): +def start_slack(bot_token, channel_id="", poll_interval=60): global _instance _instance = SlackChannel(bot_token, channel_id, poll_interval) - return _instance.start(auth_secret) + return _instance.start() def stop_slack(): if _instance: diff --git a/channels/telegram.py b/channels/telegram.py index dbb7f020..a52bd0ed 100644 --- a/channels/telegram.py +++ b/channels/telegram.py @@ -4,6 +4,7 @@ import urllib.parse import urllib.request +import auth from channels.base import BaseChannel class TelegramChannel(BaseChannel): @@ -19,30 +20,25 @@ def __init__(self, bot_token, chat_id="", poll_timeout=20): self._state_lock = threading.Lock() self._authenticated_chat_id = None - def _set_auth_secret(self, secret=None) -> None: - super()._set_auth_secret(secret) - with self._auth_lock: - self._authenticated_chat_id = None - def _is_allowed_message(self, sender_id: str, msg: str, chat_id: str = "") -> str: - candidate = self._parse_auth_candidate(msg) with self._auth_lock: if self._chat_id and chat_id != self._chat_id: return "ignore" - if not self._auth_secret: + if not auth.is_auth_enabled(): if not self._chat_id: self._chat_id = chat_id return "allow" - if self._authenticated_id is None: - if candidate == self._auth_secret: - self._authenticated_id = sender_id - self._authenticated_chat_id = chat_id - self._chat_id = chat_id - return "auth_bound" - return "ignore" - if chat_id != self._authenticated_chat_id: - return "ignore" - return "allow" if sender_id == self._authenticated_id else "ignore" + if self._authenticated_id is not None: + if chat_id != self._chat_id: + return "ignore" + return "allow" if sender_id == self._authenticated_id else "ignore" + candidate = self._parse_auth_candidate(msg) + if auth.verify_token(candidate): + self._authenticated_id = sender_id + self._authenticated_chat_id = chat_id + self._chat_id = chat_id + return "auth_bound" + return "ignore" def _api_call(self, method, params=None, timeout=30, use_post=False): params = params or {} @@ -141,18 +137,21 @@ def send_message(self, text: str) -> None: print(f"[TELEGRAM] Send failed: {exc}") return - def start(self, auth_secret=None) -> threading.Thread: - self._set_auth_secret(auth_secret) + def start(self) -> threading.Thread: + proxy = auth.get_proxy_url() + if proxy: + self._bot_token = "proxy" + self._api_base = f"{proxy}/telegram" print(f"[TELEGRAM] Starting adapter with chat target: {self._chat_id or 'auto-bind'}") self._initialize_offset() - return super().start(auth_secret=None) # secret already set above + return super().start() _instance: TelegramChannel | None = None -def start_telegram(bot_token, chat_id="", poll_timeout=20, auth_secret=None): +def start_telegram(bot_token, chat_id="", poll_timeout=20): global _instance _instance = TelegramChannel(bot_token, chat_id, poll_timeout) - return _instance.start(auth_secret) + return _instance.start() def stop_telegram(): if _instance: From 21760497db7dc0fea3df5f27124ed5a328e2a015 Mon Sep 17 00:00:00 2001 From: Bereket-Eshete Date: Thu, 25 Jun 2026 11:11:01 +0300 Subject: [PATCH 16/16] fix: restore auth gate and fix proxy token order --- channels/base.py | 7 +++++++ channels/channels_registry.py | 4 ++-- channels/irc.py | 4 +++- channels/mattermost.py | 2 +- channels/slack.py | 34 +++++++++++++++++++--------------- channels/telegram.py | 8 +++++--- 6 files changed, 37 insertions(+), 22 deletions(-) diff --git a/channels/base.py b/channels/base.py index fc8df576..d2d30c51 100644 --- a/channels/base.py +++ b/channels/base.py @@ -39,12 +39,19 @@ def _parse_auth_candidate(msg: str) -> str: return text[6:].strip() return text + @staticmethod + def _is_auth_command(msg: str) -> bool: + lower = msg.strip().lower() + return lower.startswith("auth ") or lower.startswith("/auth ") + def _is_allowed_message(self, sender_id: str, msg: str) -> str: with self._auth_lock: if not auth.is_auth_enabled(): return "allow" if self._authenticated_id is not None: return "allow" if sender_id == self._authenticated_id else "ignore" + if not self._is_auth_command(msg): + return "ignore" candidate = self._parse_auth_candidate(msg) if auth.verify_token(candidate): self._authenticated_id = sender_id diff --git a/channels/channels_registry.py b/channels/channels_registry.py index 1d3474bc..18165720 100644 --- a/channels/channels_registry.py +++ b/channels/channels_registry.py @@ -7,8 +7,8 @@ lambda token, channel_id, poll_interval, server_url: ( channel_id or "##omegaclaw", server_url or "irc.quakenet.org", - 6667, - "omegaclaw", + int(os.environ.get("IRC_PORT", "6667")), + os.environ.get("IRC_USER", "omegaclaw"), ), ), "telegram": ( diff --git a/channels/irc.py b/channels/irc.py index 8b2a8c43..e538c73f 100644 --- a/channels/irc.py +++ b/channels/irc.py @@ -27,6 +27,8 @@ def _is_allowed_message(self, sender_id: str, msg: str) -> str: return "allow" if self._authenticated_id is not None: return "allow" if norm_nick == self._authenticated_id else "ignore" + if not self._is_auth_command(msg): + return "ignore" candidate = self._parse_auth_candidate(msg) if auth.verify_token(candidate): self._authenticated_id = norm_nick @@ -116,7 +118,7 @@ def _run_loop(self) -> None: print("[IRC] Disconnected") -_instance: IRCChannel | None = None +_instance = None def start_irc(channel, server="irc.quakenet.org", port=6667, nick="omegaclaw"): diff --git a/channels/mattermost.py b/channels/mattermost.py index 76fa4e67..8f738159 100644 --- a/channels/mattermost.py +++ b/channels/mattermost.py @@ -76,7 +76,7 @@ def send_message(self, text: str) -> None: requests.post(f"{self._url}/api/v4/posts", headers=self._headers, json={"channel_id": self._channel_id, "message": text}) -_instance: MattermostChannel | None = None +_instance = None def start_mattermost(url, channel_id, bot_token): global _instance diff --git a/channels/slack.py b/channels/slack.py index 09e6285f..10cccd95 100644 --- a/channels/slack.py +++ b/channels/slack.py @@ -32,16 +32,15 @@ class SlackChannel(BaseChannel): def __init__(self, bot_token, channel_id="", poll_interval=60): super().__init__() self._bot_token = str(bot_token).strip() - if not self._bot_token: - raise ValueError("SL_BOT_TOKEN is required") self._api_base = "https://slack.com" self._channel_id = str(channel_id).strip() self._poll_interval = max(1, int(poll_interval)) self._bot_user_id = "" - self._user_cache: dict = {} - self._channel_offsets: dict = {} - self._channel_name_cache: dict = {} - self._auto_bind_channels: list = [] + self._state_lock = threading.Lock() + self._user_cache = {} + self._channel_offsets = {} + self._channel_name_cache = {} + self._auto_bind_channels = [] self._auto_bind_index = 0 self._auto_bind_last_refresh = 0.0 self._rate_limit_until = 0.0 @@ -52,12 +51,15 @@ def _is_allowed_message(self, sender_id: str, msg: str, channel_id: str = "") -> return "ignore" if not auth.is_auth_enabled(): if not self._channel_id: - label = self._channel_name_cache.get(channel_id, channel_id) + with self._state_lock: + label = self._channel_name_cache.get(channel_id, channel_id) print(f"[SLACK] Auto-bound to channel {label}") self._channel_id = channel_id return "allow" if self._authenticated_id is not None: return "allow" if sender_id == self._authenticated_id else "ignore" + if not self._is_auth_command(msg): + return "ignore" candidate = self._parse_auth_candidate(msg) if auth.verify_token(candidate): self._authenticated_id = sender_id @@ -74,7 +76,7 @@ def _api_call(self, method, params=None, timeout=30): "Content-Type": "application/x-www-form-urlencoded"}, method="POST", ) - with self._auth_lock: + with self._state_lock: wait = self._rate_limit_until - time.time() if wait > 0: time.sleep(wait) @@ -85,7 +87,7 @@ def _api_call(self, method, params=None, timeout=30): except urllib.error.HTTPError as exc: if exc.code == 429: retry = max(1, int(exc.headers.get("Retry-After", 60))) - with self._auth_lock: + with self._state_lock: self._rate_limit_until = time.time() + retry raise _RateLimitError(retry) from exc raise @@ -93,14 +95,14 @@ def _api_call(self, method, params=None, timeout=30): err = payload.get("error", f"{method} failed") if err == "ratelimited": retry = max(1, int(headers.get("Retry-After", 60))) - with self._auth_lock: + with self._state_lock: self._rate_limit_until = time.time() + retry raise _RateLimitError(retry) raise RuntimeError(err) return payload def _get_display_name(self, user_id): - with self._auth_lock: + with self._state_lock: cached = self._user_cache.get(user_id) if cached: return cached @@ -112,7 +114,7 @@ def _get_display_name(self, user_id): (payload["user"].get("name")) or user_id).strip() except Exception as exc: print(f"[SLACK] Could not resolve user {user_id}: {exc}") - with self._auth_lock: + with self._state_lock: self._user_cache[user_id] = name return name @@ -202,7 +204,7 @@ def _run_loop(self) -> None: print("[SLACK] Polling started") while self._running: try: - with self._auth_lock: + with self._state_lock: bound = self._channel_id if bound: if bound not in self._channel_offsets: @@ -232,7 +234,7 @@ def send_message(self, text: str) -> None: text = str(text).replace("\\n", "\n").replace("\r", "") if not text: return - with self._auth_lock: + with self._state_lock: target = self._channel_id if not target: return @@ -249,6 +251,8 @@ def start(self) -> threading.Thread: if proxy: self._bot_token = "proxy" self._api_base = f"{proxy}/slack" + elif not self._bot_token: + raise ValueError("SL_BOT_TOKEN is required") payload = self._api_call("auth.test", timeout=15) self._bot_user_id = str(payload.get("user_id", "")).strip() if self._channel_id: @@ -264,7 +268,7 @@ def start(self) -> threading.Thread: print(f"[SLACK] Starting adapter with channel target: {self._channel_id or 'auto-bind'}") return super().start() -_instance: SlackChannel | None = None +_instance = None def start_slack(bot_token, channel_id="", poll_interval=60): global _instance diff --git a/channels/telegram.py b/channels/telegram.py index a52bd0ed..b28c5a58 100644 --- a/channels/telegram.py +++ b/channels/telegram.py @@ -11,8 +11,6 @@ class TelegramChannel(BaseChannel): def __init__(self, bot_token, chat_id="", poll_timeout=20): super().__init__() self._bot_token = str(bot_token).strip() - if not self._bot_token: - raise ValueError("TG_BOT_TOKEN is required") self._api_base = f"https://api.telegram.org/bot{self._bot_token}" self._chat_id = str(chat_id).strip() self._poll_timeout = max(1, int(poll_timeout)) @@ -32,6 +30,8 @@ def _is_allowed_message(self, sender_id: str, msg: str, chat_id: str = "") -> st if chat_id != self._chat_id: return "ignore" return "allow" if sender_id == self._authenticated_id else "ignore" + if not self._is_auth_command(msg): + return "ignore" candidate = self._parse_auth_candidate(msg) if auth.verify_token(candidate): self._authenticated_id = sender_id @@ -142,11 +142,13 @@ def start(self) -> threading.Thread: if proxy: self._bot_token = "proxy" self._api_base = f"{proxy}/telegram" + elif not self._bot_token: + raise ValueError("TG_BOT_TOKEN is required") print(f"[TELEGRAM] Starting adapter with chat target: {self._chat_id or 'auto-bind'}") self._initialize_offset() return super().start() -_instance: TelegramChannel | None = None +_instance = None def start_telegram(bot_token, chat_id="", poll_timeout=20): global _instance