Compare commits

...
Author SHA1 Message Date
whyour 31aa353d7d test(users): provide JWT dependency to polling fixtures 2026-10-12 01:18:48 +08:00
whyour aafd85cbcc fix(users): enforce read-only ownership and revocable managed sessions 2026-10-12 01:13:53 +08:00
whyour 70636900b2 chore: finalize 2.23.0 publication metadata 2026-10-11 12:27:51 +08:00
whyour 331f4f9138 Prepare Qinglong 2.23.0 and CLI 0.1.2 release (#3090)
* fix: patch reachable release-blocking dependencies

* docs: keep release risk notes in PR description

* chore: prepare Qinglong 2.23.0 and CLI 0.1.2 release metadata
2026-10-11 12:24:50 +08:00
whyour b06cb72334 fix: apply minimal production dependency security patches 2026-10-11 02:09:20 +08:00
whyour 89c6355635 fix: renew system tokens and serialize authentication cache updates 2026-10-11 01:56:50 +08:00
whyour 68ffa106f8 fix: resolve Debian regression findings (#3087)
* fix: resolve Debian regression findings

* fix: reject failed APT index refreshes and simplify SSH home lookup
2026-10-09 01:41:29 +08:00
whyour 9d5f58ef99 ci: verify Gitee Pull mirrors with resilient GitHub notifications
Use native Gitee Pull mirrors for code and static artifacts, and verify branch and tag SHAs alongside image builds. Trigger stale mirrors through small GitHub REST ref updates, retry transient errors with backoff, and keep polling when a notification fails.
2026-10-08 01:16:43 +08:00
whyour ef1c54a620 fix: 补齐前端缺失的中英文翻译 2026-10-07 20:35:59 +08:00
whyour 0218404d9f fix: 修复页面升级、ql update 与静态资源版本同步 (#3086)
* fix: validate panel upgrades and synchronize versioned static builds

* fix: add static-only mirror backfill workflow

* fix: avoid lifecycle self-kill and stop cancelled upgrade groups

* fix: build images when the shared upgrade engine changes

* fix: dispatch upgrade lifecycle through a fixed script

* fix: reclaim upgrade locks safely and normalize health probes

* fix: match backend base URL normalization exactly
2026-10-07 17:43:03 +08:00
whyour 24f7d1a46f fix: resolve dev panel and CLI regressions
Synchronize scheduler concurrency across API and gRPC processes and restore persisted limits during recovery.

Serialize subscription mutations and preserve shared storage while cleaning Shell and CLI namespaces, historical references, and filesystem-specific naming candidates.

Handle CLI text responses correctly and include script history in backups. Add regression coverage for concurrency, cleanup ownership, long URLs, Unicode paths, and history exports.
2026-10-07 12:35:10 +08:00
whyour fa8910b3d6 feat: add script version history (#3085)
* feat: add script version history

* fix: validate canonical save-as paths before backup

* fix: stack script history footer on small screens
2026-10-06 22:00:11 +08:00
whyour 2cab112391 fix: 加固安全输入边界并修复订阅兼容性 (#3084)
* fix: harden scoped API inputs and isolate login throttling

* fix: close script path gaps and harden security boundaries

* fix: preserve subscription aliases and avoid extension eval

* fix: allow cancelling invalid historical subscriptions

* fix: make script parent validation explicit for CodeQL

* fix: limit command execution and clarify trusted path branches

* fix: separate diff tool script access from config APIs

* fix: clean up obsolete subscription SSH credentials
2026-10-06 18:21:54 +08:00
whyour 2ab9c702e6 ci: skip Docker builds for CLI-only changes 2026-10-01 12:25:16 +08:00
whyour 6dd98a8565 fix(ci): isolate CLI release concurrency from validation 2026-10-01 12:17:26 +08:00
whyour 06c8d8b112 fix: report resource conflicts and accurate CLI API errors 2026-10-01 11:26:40 +08:00
whyour a5406bedca ci: allow verified CLI prereleases from develop 2026-10-01 00:26:42 +08:00
whyour 3d2b98a442 fix: scope file access restrictions and honor custom data directories 2026-09-30 21:21:35 +08:00
whyour 93d7ec69dc fix: scope config file blacklist to config directory (#3082) 2026-09-30 20:31:20 +08:00
whyour a303e33b4e fix: invalidate queued cron revisions and enforce task process timeouts 2026-09-29 23:37:00 +08:00
whyour 44129ca088 fix: align CLI shell completion and cancellation semantics 2026-09-29 01:06:47 +08:00
whyour def4917447 fix: finalize exited shell tasks and preserve command environments 2026-09-29 00:52:33 +08:00
whyour bc0f35e3f7 fix: reduce task startup overhead and lazy-load notifications (#3080) 2026-09-28 23:41:31 +08:00
whyour f168efcdae refactor: migrate cron scheduling to node-cron (#3079)
* refactor: migrate cron scheduling to node-cron

* feat: support annually midnight and minutely cron macros

* fix: harden cron recovery and system scheduler compatibility
2026-09-27 22:18:39 +08:00
whyour bf63425805 fix: isolate invalid cron rules during scheduler recovery 2026-09-27 16:45:09 +08:00
whyour a8310e7d51 fix: restore legacy cron schedules and shell diagnostics (#3078) 2026-09-27 16:16:22 +08:00
whyour a67d44cde5 fix: pin Debian Python 3.10 base for full release matrix 2026-09-26 01:40:39 +08:00
whyour 6c487018c2 fix: drain completed task logs and prepare 2.22.0 release (#3076) 2026-09-26 01:17:28 +08:00
whyour 801a71d740 feat(cli): cover OpenAPI with a remote npm CLI and internal panel tools (#3074)
* feat(cli): add unified Commander CLI for QingLong 2.x

* fix(cli): publish via npm and address security review feedback

* ci(cli): package npm artifacts and remove evaluation collateral

* test(cli): use a fixed shell fixture for log retention

* refactor(cli): separate remote npm client from panel tools

* feat(cli): cover active panel OpenAPI resources

* docs(cli): unify authentication and skill guidance

* refactor(cli): isolate internal commands and generate Commander help

* refactor(cli): organize remote and internal modules by responsibility

* ci(cli): publish verified npm archives from master

* fix(cli): publish under the whyour npm scope

* ci: use npm trusted publishing for both packages

* docs: introduce the published CLI on the project homepage

* fix(cli): preserve server log truncation and correct login hints

* fix(cli): accept dashboard record request bodies

* fix(cli): preserve stdin for local task execution

* fix(cli): resolve task executables after changing directory

* fix(cli): preserve shell function tasks and sanitize test failures

* fix(cli): preserve shell hook state and resolve workdir after hooks

* fix(cli): preserve cleanup across shared shell task timeouts

* fix(cli): isolate shell control descriptors and reap timed-out descendants
2026-09-25 23:24:41 +08:00
whyour f051135fc4 fix: declare nullable subscription process id 2026-09-24 20:37:13 +08:00
whyour 2d6c2206b3 fix: verify task termination and preserve startup configuration 2026-09-22 22:23:18 +08:00
whyour 6acb665369 fix: prevent expired-token login refresh loop 2026-09-19 13:23:28 +08:00
whyour d9833c17d5 perf: reduce resident dependencies and duplicate WebSocket authentication (#3071)
* perf: load HTTP dependencies only in the HTTP worker

* docs: record 30-minute idle resource comparison

* perf: isolate database initialization from the cluster primary

* docs: record isolated database startup resource comparison

* perf: avoid an extra shell child in Docker health probes

* perf: load notification dependencies only when needed

* perf: retain queue cleanup after notification loading comparison

* fix: stop page polling after unmount and avoid duplicate fetches

* perf: share WebSocket session polling across connections

* docs: measure real WebSocket container CPU and memory

* perf: reuse JWT validation within each WebSocket session check

* perf: bound login IP lookup allocations with an address cache

* refactor: limit resource optimization PR to three core changes

* docs: benchmark the three core optimizations against develop

* chore: remove documentation and benchmark artifacts from PR
2026-09-18 20:22:18 +08:00
whyour a60ec5587a feat: show today's successful tasks from dashboard (#3072) 2026-09-18 02:16:54 +08:00
whyour 4eb27427f8 fix: 修复 develop 调度锁、运行恢复与 Alpine 缓存并发 (#3070)
* fix: repair develop scheduler locks and runtime recovery

* test: wait for child reaping after verified termination

* fix: abort cron mutations when scheduler deletion fails
2026-09-13 13:44:44 +08:00
whyour 041a437453 fix: support WPUSH notifications over gRPC 2026-09-13 01:08:14 +08:00
Alone88 841740f19a feat: add WPUSH notification provider (#3066)
Add WPUSH as a notification channel for Qinglong panel settings and
sample notify scripts (Python/JS), using POST https://api.wpush.cn/api/v1/send
with success code === 0. Supports optional channel and topic_code.
2026-09-13 00:50:56 +08:00
whyour a94e665054 fix: 修复任务生命周期与调度就绪,优化执行和构建开销 (#3069)
* fix: harden task lifecycle and scheduler readiness

* fix: confine log writes to the configured log directory

* fix: verify complete build artifacts and untracked inputs

* fix: reconcile scheduler state and make stop win startup races

* fix: isolate cron generations and serialize scheduler recovery
2026-09-13 00:32:41 +08:00
whyour d62d8f3025 fix: support Chinese cron log names (#3068) 2026-09-12 03:03:25 +08:00
copilot-swe-agent[bot]andwhyour 0deebcfc88 Hide system log and login log tabs for non-admin users
Co-authored-by: whyour <22700758+whyour@users.noreply.github.com>
2025-11-11 16:53:19 +00:00
copilot-swe-agent[bot]andwhyour 07fcb09cc6 Add log isolation and admin-only access for system/login logs
Co-authored-by: whyour <22700758+whyour@users.noreply.github.com>
2025-11-11 16:36:12 +00:00
copilot-swe-agent[bot]andwhyour 6aefc61be6 Fix authentication for regular users by validating JWT tokens
Co-authored-by: whyour <22700758+whyour@users.noreply.github.com>
2025-11-10 18:10:16 +00:00
copilot-swe-agent[bot]andwhyour bf9be821ba Add ALTER TABLE statements for userId columns in db.ts
Co-authored-by: whyour <22700758+whyour@users.noreply.github.com>
2025-11-10 16:57:18 +00:00
copilot-swe-agent[bot]andwhyour d42074f76a Add data migration script and comprehensive migration guide
Co-authored-by: whyour <22700758+whyour@users.noreply.github.com>
2025-11-10 16:48:31 +00:00
copilot-swe-agent[bot]andwhyour 5c798a0e93 Add user management frontend interface for admins
Co-authored-by: whyour <22700758+whyour@users.noreply.github.com>
2025-11-10 16:38:26 +00:00
whyour 15d94f469b Merge branch 'develop' into copilot/enable-multi-user-management 2025-11-10 23:53:10 +08:00
copilot-swe-agent[bot]andwhyour 777fd3fb23 Add user-scoped data filtering for subscription and dependence operations
Co-authored-by: whyour <22700758+whyour@users.noreply.github.com>
2025-11-09 16:37:37 +00:00
whyour 4cf2858ab0 Merge branch 'develop' into copilot/enable-multi-user-management 2025-11-09 19:50:16 +08:00
copilot-swe-agent[bot]andwhyour 2ff7a186e7 Changes before error encountered
Co-authored-by: whyour <22700758+whyour@users.noreply.github.com>
2025-11-07 16:38:20 +00:00
copilot-swe-agent[bot]andwhyour b2b1777c6b Security improvements: Fix ownership checks, add password hashing with bcrypt
Co-authored-by: whyour <22700758+whyour@users.noreply.github.com>
2025-11-07 16:37:05 +00:00
copilot-swe-agent[bot]andwhyour f355b4e441 Add user-scoped data filtering for env operations
Co-authored-by: whyour <22700758+whyour@users.noreply.github.com>
2025-11-07 16:33:31 +00:00
copilot-swe-agent[bot]andwhyour 489454daa0 Add user-scoped data filtering for cron operations
Co-authored-by: whyour <22700758+whyour@users.noreply.github.com>
2025-11-07 16:30:04 +00:00
copilot-swe-agent[bot]andwhyour db93ca9aa9 Add multi-user backend infrastructure: User model, management service and API
Co-authored-by: whyour <22700758+whyour@users.noreply.github.com>
2025-11-07 16:24:00 +00:00
copilot-swe-agent[bot] 4758400df6 Initial plan 2025-11-07 16:12:26 +00:00
414 changed files with 45498 additions and 1696 deletions
+13
View File
@@ -0,0 +1,13 @@
.git
.gitnexus
.claude
.codex
.env*
node_modules
data
audit
docs
test
src/.umi*
**/__pycache__
**/*.log
+173 -79
View File
@@ -2,20 +2,44 @@ name: Build And Push Docker Image
on:
push:
paths-ignore:
- "*.md"
paths:
- "**"
- "!*.md"
# CLI is verified separately; its shared Bash upgrade engine ships in images.
- "!cli/**"
- "cli/src/internal/maintenance/upgradeArtifacts.cjs"
- "!.github/workflows/cli-package.yml"
branches:
- "master"
- "develop"
tags:
- "v*"
workflow_dispatch:
inputs:
static_sync_only:
description: Synchronize an existing static snapshot to GitLab and verify Gitee
type: boolean
default: false
static_ref:
description: Static branch or version name, for example v2.22.0
type: string
default: ''
static_ref_type:
description: Synchronize a branch, or a version branch and Git tag
type: choice
options: [branch, tag]
default: branch
permissions:
contents: read
jobs:
validate:
if: ${{ !inputs.static_sync_only }}
uses: ./.github/workflows/validate.yml
code_gitlab:
if: ${{ !inputs.static_sync_only }}
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v6
@@ -32,37 +56,9 @@ jobs:
git push --force --all gitlab 2>&1 || echo "::warning::GitLab push failed"
git push --force --tags gitlab 2>&1 || echo "::warning::GitLab tags push failed"
code_gitee:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v6
with:
fetch-depth: 0
- name: Push to Gitee
env:
GITEE_TOKEN: ${{ secrets.GITEE_TOKEN }}
run: |
set +e
mkdir -p ~/.ssh
printf '%s\n' "${{ secrets.GITLAB_SSH_PK }}" > ~/.ssh/id_rsa
chmod 600 ~/.ssh/id_rsa
ssh-keyscan -T 10 gitee.com >> ~/.ssh/known_hosts 2>/dev/null
git remote add gitee git@gitee.com:whyour/qinglong.git 2>/dev/null
if git push --force --all gitee 2>&1; then
echo "::notice::Gitee push --all succeeded"
else
echo "::warning::Push failed, trying to create repo via API..."
curl -sS --connect-timeout 30 --max-time 60 \
-X POST "https://gitee.com/api/v5/user/repos" \
-H "Content-Type: application/json" \
-d '{"name":"qinglong","private":"false"}' \
"?access_token=$GITEE_TOKEN" 2>/dev/null
git push --force --all gitee 2>&1 && echo "::notice::Gitee push succeeded after repo creation" || echo "::warning::Gitee push failed after retry"
fi
git push --force --tags gitee 2>&1 || echo "::warning::Gitee tags push failed"
build-static:
if: ${{ !inputs.static_sync_only }}
needs: validate
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v6
@@ -72,6 +68,7 @@ jobs:
- uses: actions/setup-node@v6
with:
node-version: "20"
cache: "pnpm"
cache-dependency-path: pnpm-lock.yaml
@@ -80,67 +77,122 @@ jobs:
pnpm install --frozen-lockfile
pnpm build:front
pnpm build:back
pnpm build:info
- uses: actions/upload-artifact@v6
with:
name: qinglong-static-${{ github.sha }}
path: static/
if-no-files-found: error
include-hidden-files: true
- name: copy to static repo
env:
GITHUB_REPO: github.com/${{ github.repository_owner }}/qinglong-static
GITHUB_BRANCH: ${{ github.ref_name }}
GITHUB_REF_TYPE: ${{ github.ref_type }}
run: |
mkdir -p tmp
cd ./tmp
cp -rf ../static/* ./
git init -b ${GITHUB_BRANCH} && git add .
git config --local user.name 'github-actions[bot]'
git config --local user.email 'github-actions[bot]@users.noreply.github.com'
git commit --allow-empty -m "copy static at $(date +'%Y-%m-%d %H:%M:%S')"
git push --force --quiet "https://${{ secrets.API_TOKEN }}@${GITHUB_REPO}.git" ${GITHUB_BRANCH}:${GITHUB_BRANCH}
bash scripts/publish-static.sh "https://${{ secrets.API_TOKEN }}@${GITHUB_REPO}.git" "$GITHUB_BRANCH" "$GITHUB_REF_TYPE"
static_gitlab:
needs: build-static
runs-on: ubuntu-latest
concurrency:
group: static-gitlab-${{ github.ref_name }}
cancel-in-progress: false
steps:
- uses: actions/checkout@v6
- name: Push qinglong-static to GitLab
env:
STATIC_BRANCH: ${{ github.ref_name }}
STATIC_REF_TYPE: ${{ github.ref_type }}
MIRROR_SSH_KEY: ${{ secrets.GITLAB_SSH_PK }}
run: |
set +e
set -euo pipefail
mkdir -p ~/.ssh
printf '%s\n' "${{ secrets.GITLAB_SSH_PK }}" > ~/.ssh/id_rsa
printf '%s\n' "$MIRROR_SSH_KEY" > ~/.ssh/id_rsa
chmod 600 ~/.ssh/id_rsa
ssh-keyscan -T 10 gitlab.com >> ~/.ssh/known_hosts 2>/dev/null
git clone --depth=1 --single-branch https://github.com/whyour/qinglong-static.git static-mirror
cd static-mirror
git remote add gitlab git@gitlab.com:whyour/qinglong-static.git 2>/dev/null
git push --force --all gitlab 2>&1 || echo "::warning::GitLab static push failed"
git push --force --tags gitlab 2>&1 || echo "::warning::GitLab static tags push failed"
bash scripts/sync-static.sh git@gitlab.com:whyour/qinglong-static.git "$STATIC_BRANCH" "$STATIC_REF_TYPE"
static_gitee:
verify-gitee-mirrors:
needs: build-static
runs-on: ubuntu-latest
timeout-minutes: 14
permissions:
contents: write
strategy:
fail-fast: false
matrix:
repository: [qinglong, qinglong-static]
steps:
- name: Push qinglong-static to Gitee
- uses: actions/checkout@v6
# Gitee Pull mirrors and GitHub push webhooks own synchronization.
# Verification runs alongside Docker builds and needs no Gitee credentials.
- name: Wait for Gitee Pull mirror
if: ${{ matrix.repository == 'qinglong' }}
env:
GITEE_TOKEN: ${{ secrets.GITEE_TOKEN }}
GH_TOKEN: ${{ github.token }}
QL_MIRROR_NOTIFY_REPO: whyour/qinglong
MIRROR_REPOSITORY: ${{ matrix.repository }}
MIRROR_REF: ${{ github.ref_name }}
MIRROR_REF_TYPE: ${{ github.ref_type }}
QL_MIRROR_INCLUDE_VERSION_BRANCH: ${{ matrix.repository == 'qinglong-static' }}
QL_MIRROR_NOTIFY_ON_START: 'false'
run: |
set +e
mkdir -p ~/.ssh
printf '%s\n' "${{ secrets.GITLAB_SSH_PK }}" > ~/.ssh/id_rsa
chmod 600 ~/.ssh/id_rsa
ssh-keyscan -T 10 gitee.com >> ~/.ssh/known_hosts 2>/dev/null
git clone --depth=1 --single-branch https://github.com/whyour/qinglong-static.git static-mirror
cd static-mirror
git remote add gitee git@gitee.com:whyour/qinglong-static.git 2>/dev/null
bash scripts/wait-for-mirror.sh "https://github.com/whyour/$MIRROR_REPOSITORY.git" "https://gitee.com/whyour/$MIRROR_REPOSITORY.git" "$MIRROR_REF" "$MIRROR_REF_TYPE"
- name: Notify and wait for Gitee static mirror
if: ${{ matrix.repository == 'qinglong-static' }}
env:
GH_TOKEN: ${{ secrets.API_TOKEN }}
QL_MIRROR_NOTIFY_REPO: whyour/qinglong-static
MIRROR_REF: ${{ github.ref_name }}
MIRROR_REF_TYPE: ${{ github.ref_type }}
QL_MIRROR_INCLUDE_VERSION_BRANCH: 'true'
run: |
bash scripts/wait-for-mirror.sh https://github.com/whyour/qinglong-static.git https://gitee.com/whyour/qinglong-static.git "$MIRROR_REF" "$MIRROR_REF_TYPE"
if git push --force --all gitee 2>&1; then
echo "::notice::Gitee static push succeeded"
else
echo "::warning::Push failed, trying to create repo via API..."
curl -sS --connect-timeout 30 --max-time 60 \
-X POST "https://gitee.com/api/v5/user/repos" \
-H "Content-Type: application/json" \
-d '{"name":"qinglong-static","private":"false"}' \
"?access_token=$GITEE_TOKEN" 2>/dev/null
git push --force --all gitee 2>&1 && echo "::notice::Gitee static push succeeded after repo creation" || echo "::warning::Gitee static push failed after retry"
fi
git push --force --tags gitee 2>&1 || echo "::warning::Gitee static tags push failed"
sync-existing-static:
if: ${{ github.event_name == 'workflow_dispatch' && inputs.static_sync_only }}
runs-on: ubuntu-latest
strategy:
fail-fast: false
matrix:
mirror: [gitlab]
concurrency:
group: static-${{ matrix.mirror }}-${{ inputs.static_ref }}
cancel-in-progress: false
steps:
- uses: actions/checkout@v6
- name: Synchronize existing static snapshot
env:
STATIC_REF: ${{ inputs.static_ref }}
STATIC_REF_TYPE: ${{ inputs.static_ref_type }}
MIRROR_HOST: ${{ matrix.mirror }}.com
MIRROR_SSH_KEY: ${{ secrets.GITLAB_SSH_PK }}
run: |
set -euo pipefail
mkdir -p ~/.ssh
printf '%s\n' "$MIRROR_SSH_KEY" > ~/.ssh/id_rsa
chmod 600 ~/.ssh/id_rsa
ssh-keyscan -T 10 "$MIRROR_HOST" >> ~/.ssh/known_hosts 2>/dev/null
bash scripts/sync-static.sh "git@$MIRROR_HOST:whyour/qinglong-static.git" "$STATIC_REF" "$STATIC_REF_TYPE"
verify-existing-static-gitee:
if: ${{ github.event_name == 'workflow_dispatch' && inputs.static_sync_only }}
runs-on: ubuntu-latest
timeout-minutes: 14
steps:
- uses: actions/checkout@v6
- name: Wait for existing Gitee static snapshot
env:
GH_TOKEN: ${{ secrets.API_TOKEN }}
QL_MIRROR_NOTIFY_REPO: whyour/qinglong-static
MIRROR_REF: ${{ inputs.static_ref }}
MIRROR_REF_TYPE: ${{ inputs.static_ref_type }}
QL_MIRROR_INCLUDE_VERSION_BRANCH: 'true'
run: |
bash scripts/wait-for-mirror.sh https://github.com/whyour/qinglong-static.git https://gitee.com/whyour/qinglong-static.git "$MIRROR_REF" "$MIRROR_REF_TYPE"
build-alpine:
if: ${{ !startsWith(github.ref, 'refs/tags/') }}
@@ -151,11 +203,16 @@ jobs:
contents: read
steps:
- uses: actions/checkout@v6
- uses: actions/download-artifact@v7
with:
name: qinglong-static-${{ github.sha }}
path: static/
- uses: pnpm/action-setup@v6
with:
version: "8.3.1"
- uses: actions/setup-node@v6
with:
node-version: "20"
cache: "pnpm"
cache-dependency-path: pnpm-lock.yaml
@@ -208,7 +265,7 @@ jobs:
uses: docker/build-push-action@v7
with:
build-args: |
MAINTAINER=${{ github.repository_owner }}
QL_MAINTAINER=${{ github.repository_owner }}
QL_BRANCH=${{ github.ref_name }}
SOURCE_COMMIT=${{ github.sha }}
network: host
@@ -232,11 +289,16 @@ jobs:
contents: read
steps:
- uses: actions/checkout@v6
- uses: actions/download-artifact@v7
with:
name: qinglong-static-${{ github.sha }}
path: static/
- uses: pnpm/action-setup@v6
with:
version: "8.3.1"
- uses: actions/setup-node@v6
with:
node-version: "20"
cache: "pnpm"
cache-dependency-path: pnpm-lock.yaml
@@ -288,7 +350,7 @@ jobs:
uses: docker/build-push-action@v7
with:
build-args: |
MAINTAINER=${{ github.repository_owner }}
QL_MAINTAINER=${{ github.repository_owner }}
QL_BRANCH=${{ github.ref_name }}
SOURCE_COMMIT=${{ github.sha }}
network: host
@@ -313,11 +375,16 @@ jobs:
contents: read
steps:
- uses: actions/checkout@v6
- uses: actions/download-artifact@v7
with:
name: qinglong-static-${{ github.sha }}
path: static/
- uses: pnpm/action-setup@v6
with:
version: "8.3.1"
- uses: actions/setup-node@v6
with:
node-version: "20"
cache: "pnpm"
cache-dependency-path: pnpm-lock.yaml
@@ -355,7 +422,7 @@ jobs:
uses: docker/build-push-action@v7
with:
build-args: |
MAINTAINER=${{ github.repository_owner }}
QL_MAINTAINER=${{ github.repository_owner }}
QL_BRANCH=${{ github.ref_name }}
SOURCE_COMMIT=${{ github.sha }}
network: host
@@ -380,11 +447,16 @@ jobs:
contents: read
steps:
- uses: actions/checkout@v6
- uses: actions/download-artifact@v7
with:
name: qinglong-static-${{ github.sha }}
path: static/
- uses: pnpm/action-setup@v6
with:
version: "8.3.1"
- uses: actions/setup-node@v6
with:
node-version: "20"
cache: "pnpm"
cache-dependency-path: pnpm-lock.yaml
@@ -422,7 +494,7 @@ jobs:
uses: docker/build-push-action@v7
with:
build-args: |
MAINTAINER=${{ github.repository_owner }}
QL_MAINTAINER=${{ github.repository_owner }}
QL_BRANCH=${{ github.ref_name }}
SOURCE_COMMIT=${{ github.sha }}
network: host
@@ -440,9 +512,12 @@ jobs:
cache-to: type=registry,ref=whyour/qinglong:cache-debian-python3.10,mode=max
publish:
if: ${{ github.ref_name == 'master' }}
needs: [build-alpine, build-debian]
if: ${{ github.repository == 'whyour/qinglong' && github.ref == 'refs/heads/master' }}
needs: [build-alpine, build-debian, build-alpine310, build-debian310]
runs-on: ubuntu-latest
permissions:
contents: read
id-token: write
steps:
- uses: actions/checkout@v4
- uses: pnpm/action-setup@v3
@@ -451,6 +526,7 @@ jobs:
- uses: actions/setup-node@v3
with:
node-version: "20"
cache: "pnpm"
- name: build front and back
@@ -459,7 +535,25 @@ jobs:
pnpm build:front
pnpm build:back
- name: publish npm package
- name: Set up Node for trusted publishing
uses: actions/setup-node@v6
with:
node-version: '24'
registry-url: https://registry.npmjs.org
package-manager-cache: false
- name: Publish npm package with OIDC
shell: bash
run: |
echo "//registry.npmjs.org/:_authToken=${{ secrets.NPM_TOKEN }}" >> ~/.npmrc
npm publish
set -euo pipefail
name=$(node -p 'require("./package.json").name')
version=$(node -p 'require("./package.json").version')
if npm view "$name@$version" version --json --registry=https://registry.npmjs.org > "$RUNNER_TEMP/npm-version.json" 2> "$RUNNER_TEMP/npm-version-error.log"; then
jq -e --arg version "$version" '. == $version' "$RUNNER_TEMP/npm-version.json" > /dev/null
echo "::notice::$name@$version is already published; skipping npm publication for this image rebuild."
elif jq -e '.error.code == "E404"' "$RUNNER_TEMP/npm-version.json" > /dev/null; then
npm publish --access public --registry=https://registry.npmjs.org
else
echo '::error::Could not check the published npm version; refusing to publish.'
exit 1
fi
+161
View File
@@ -0,0 +1,161 @@
name: CLI package
on:
pull_request:
paths:
- 'cli/**'
- 'shell/**'
- 'back/api/**'
- 'back/loaders/deps.ts'
- 'package.json'
- '.github/workflows/cli-package.yml'
push:
branches: [develop, master]
paths:
- 'cli/**'
- 'shell/**'
- 'back/api/**'
- 'back/loaders/deps.ts'
- 'package.json'
- '.github/workflows/cli-package.yml'
workflow_dispatch:
inputs:
publish:
description: Publish the verified CLI package (master stable / develop prerelease)
type: boolean
default: false
prerelease_version:
description: Develop only, e.g. 0.1.2-beta.0 (alpha/beta/rc)
type: string
default: ''
permissions:
contents: read
concurrency:
# Keep validation runs separate from releases; never interrupt a running release.
# Master push releases share the same group as manually requested releases.
group: >-
cli-package-${{ github.workflow }}-${{ github.ref }}-${{
((github.event_name == 'workflow_dispatch' && inputs.publish) ||
(github.event_name == 'push' && github.ref == 'refs/heads/master')) && 'publish' || 'check' }}
cancel-in-progress: >-
${{ github.ref != 'refs/heads/master' &&
!(github.event_name == 'workflow_dispatch' && inputs.publish) }}
jobs:
package:
runs-on: ubuntu-latest
timeout-minutes: 15
strategy:
fail-fast: false
matrix:
node: ['22.12.0', '24']
steps:
- uses: actions/checkout@v6
- uses: actions/setup-node@v6
with:
node-version: ${{ matrix.node }}
cache: npm
cache-dependency-path: cli/package-lock.json
- name: Install test interpreters and tools
run: |
sudo apt-get update
sudo apt-get install -y --no-install-recommends bash ca-certificates curl git jq perl procps python3 unzip zip
npm install --global --ignore-scripts --no-audit --no-fund ts-node@10.9.2 typescript@5.2.2
- name: Set requested prerelease version before verification
if: github.event_name == 'workflow_dispatch' && inputs.prerelease_version != ''
working-directory: cli
env:
PRERELEASE_VERSION: ${{ inputs.prerelease_version }}
shell: bash
run: |
set -euo pipefail
if [[ "$GITHUB_REF" != refs/heads/develop ||
! "$PRERELEASE_VERSION" =~ ^[0-9]+\.[0-9]+\.[0-9]+-(alpha|beta|rc)\.[0-9]+$ ]]; then
echo '::error::Prereleases require develop and an explicit alpha/beta/rc version.'
exit 1
fi
npm version "$PRERELEASE_VERSION" --no-git-tag-version --ignore-scripts
- run: npm ci --prefix cli --no-audit --no-fund
- run: npm run check:cli
- run: npm run test:cli
- name: Verify offline package installation
env:
QL_CLI_PACKAGE_OUTPUT: ${{ runner.temp }}/cli-package
run: node cli/scripts/verify-package.cjs
- name: Upload verified npm archive
if: matrix.node == '24'
uses: actions/upload-artifact@v6
with:
name: qinglong-cli-${{ github.sha }}
path: ${{ runner.temp }}/cli-package/*.tgz
if-no-files-found: error
retention-days: 14
publish:
needs: package
if: >-
github.repository == 'whyour/qinglong' &&
((github.ref == 'refs/heads/master' &&
(github.event_name == 'push' ||
(github.event_name == 'workflow_dispatch' && inputs.publish))) ||
(github.ref == 'refs/heads/develop' && github.event_name == 'workflow_dispatch' &&
inputs.publish && inputs.prerelease_version != ''))
runs-on: ubuntu-latest
timeout-minutes: 5
permissions:
contents: read
id-token: write
steps:
- uses: actions/setup-node@v6
with:
node-version: '24'
package-manager-cache: false
registry-url: https://registry.npmjs.org
- uses: actions/download-artifact@v7
with:
name: qinglong-cli-${{ github.sha }}
path: cli-package
- name: Publish verified npm archive
env:
PRERELEASE_VERSION: ${{ inputs.prerelease_version }}
shell: bash
run: |
set -euo pipefail
shopt -s nullglob
archives=(cli-package/*.tgz)
if [ "${#archives[@]}" -ne 1 ]; then
echo '::error::Expected exactly one verified CLI archive.'
exit 1
fi
archive="${archives[0]}"
metadata=$(tar -xOf "$archive" package/package.json)
name=$(jq -er '.name | select(. == "@whyour/qinglong-cli")' <<< "$metadata")
version=$(jq -er '.version | select(type == "string")' <<< "$metadata")
if [[ "$GITHUB_REF" == refs/heads/develop ]]; then
if [[ "$version" != "$PRERELEASE_VERSION" ||
! "$version" =~ ^[0-9]+\.[0-9]+\.[0-9]+-(alpha|beta|rc)\.[0-9]+$ ]]; then
echo '::error::Archive version does not match the requested prerelease.'
exit 1
fi
tag="${BASH_REMATCH[1]}"
elif [[ "$GITHUB_REF" == refs/heads/master && "$version" =~ ^[0-9]+\.[0-9]+\.[0-9]+$ ]]; then
tag=latest
else
echo '::error::Refusing to publish this branch/version combination.'
exit 1
fi
if npm view "$name@$version" version --json --registry=https://registry.npmjs.org > version.json 2> version-error.log; then
jq -e --arg version "$version" '. == $version' version.json > /dev/null
if [[ "$tag" != latest ]]; then
echo '::error::Prerelease version already exists; choose a new version to test this build.'
exit 1
fi
echo "::notice::$name@$version is already published; bump cli/package.json and its lockfile to release changes."
exit 0
elif ! jq -e '.error.code == "E404"' version.json > /dev/null; then
echo '::error::Could not check the published CLI version; refusing to publish.'
exit 1
fi
npm publish "./$archive" --access public --tag "$tag" --ignore-scripts --registry=https://registry.npmjs.org
+136
View File
@@ -0,0 +1,136 @@
name: Release readiness
on:
pull_request:
paths:
- '.github/workflows/release-readiness.yml'
- 'package.json'
- 'pnpm-lock.yaml'
workflow_dispatch:
permissions:
contents: read
concurrency:
group: release-readiness-${{ github.event.pull_request.number || github.ref }}
cancel-in-progress: true
jobs:
regression:
uses: ./.github/workflows/validate.yml
static:
runs-on: ubuntu-latest
timeout-minutes: 15
steps:
- uses: actions/checkout@v6
with:
ref: ${{ github.event.pull_request.head.sha || github.sha }}
- uses: pnpm/action-setup@v6
with:
version: '8.3.1'
- uses: actions/setup-node@v6
with:
node-version: '20'
cache: pnpm
- run: pnpm install --frozen-lockfile
- run: pnpm build:front && pnpm build:back && pnpm build:info
- uses: actions/upload-artifact@v6
with:
name: release-readiness-static
path: static/
include-hidden-files: true
if-no-files-found: error
images:
needs: [regression, static]
runs-on: ubuntu-22.04
timeout-minutes: 60
strategy:
fail-fast: false
matrix:
include:
- image: alpine
file: docker/Dockerfile
cache: alpine
platforms: linux/amd64,linux/arm/v6,linux/arm/v7,linux/arm64,linux/ppc64le,linux/s390x,linux/386
- image: debian
file: docker/Dockerfile.debian
cache: debian
platforms: linux/amd64,linux/arm/v7,linux/arm64,linux/ppc64le,linux/s390x
- image: alpine-python3.10
file: docker/Dockerfile.310
cache: alpine-python3.10
platforms: linux/amd64,linux/arm/v6,linux/arm/v7,linux/arm64,linux/ppc64le,linux/s390x,linux/386
- image: debian-python3.10
file: docker/Dockerfile.debian310
cache: debian-python3.10
platforms: linux/amd64,linux/arm/v7,linux/arm64,linux/ppc64le,linux/s390x
steps:
- uses: actions/checkout@v6
with:
ref: ${{ github.event.pull_request.head.sha || github.sha }}
- name: Record source
id: source
env:
SOURCE_URL: ${{ github.event.pull_request.head.repo.clone_url || format('https://github.com/{0}.git', github.repository) }}
SOURCE_BRANCH: ${{ github.head_ref || github.ref_name }}
run: |
echo "sha=$(git rev-parse HEAD)" >> "$GITHUB_OUTPUT"
echo "url=$SOURCE_URL" >> "$GITHUB_OUTPUT"
echo "branch=$SOURCE_BRANCH" >> "$GITHUB_OUTPUT"
- uses: actions/download-artifact@v7
with:
name: release-readiness-static
path: static/
- uses: docker/setup-qemu-action@v4
with:
cache-image: false
- uses: docker/setup-buildx-action@v4
- name: Build every release architecture
uses: docker/build-push-action@v7
with:
context: .
file: ${{ matrix.file }}
platforms: ${{ matrix.platforms }}
build-args: |
QL_URL=${{ steps.source.outputs.url }}
QL_BRANCH=${{ steps.source.outputs.branch }}
SOURCE_COMMIT=${{ steps.source.outputs.sha }}
push: false
provenance: false
sbom: false
cache-from: type=registry,ref=whyour/qinglong:cache-${{ matrix.cache }}
- name: Load native image for runtime verification
uses: docker/build-push-action@v7
with:
context: .
file: ${{ matrix.file }}
platforms: linux/amd64
build-args: |
QL_URL=${{ steps.source.outputs.url }}
QL_BRANCH=${{ steps.source.outputs.branch }}
SOURCE_COMMIT=${{ steps.source.outputs.sha }}
tags: qinglong-readiness:${{ matrix.image }}
load: true
push: false
provenance: false
sbom: false
- name: Verify native production runtime without external networking
run: |
docker run --rm --network none --entrypoint node \
qinglong-readiness:${{ matrix.image }} \
-e 'const assert=require("node:assert/strict"); const {createRequire}=require("node:module"); for(const [name,want] of Object.entries({"@grpc/grpc-js":"1.14.5",compression:"1.8.2",multer:"2.3.0",sequelize:"6.37.8"}))assert.equal(require(name+"/package.json").version,want); const expressRequire=createRequire(require.resolve("express")); assert.equal(expressRequire("proxy-addr/package.json").version,"2.0.8"); const sockRequire=createRequire(require.resolve("sockjs")); assert.equal(sockRequire("websocket-driver/package.json").version,"0.7.5"); require("sqlite3"); require("nodemailer"); require("undici"); console.log("Minimal production patches and runtime imports verified")'
- name: Export native Debian image for isolated dev validation
if: matrix.image == 'debian'
run: |
docker save qinglong-readiness:debian | gzip -1 > release-readiness-debian-image.tar.gz
sha256sum release-readiness-debian-image.tar.gz > release-readiness-debian-image.sha256
docker image inspect qinglong-readiness:debian > release-readiness-debian-image.json
- uses: actions/upload-artifact@v6
if: matrix.image == 'debian'
with:
name: release-readiness-debian-image
path: release-readiness-debian-image.*
compression-level: 0
if-no-files-found: error
+31
View File
@@ -0,0 +1,31 @@
name: Validate
on:
pull_request:
workflow_dispatch:
workflow_call:
permissions:
contents: read
jobs:
regression:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v6
- uses: pnpm/action-setup@v6
with:
version: '8.3.1'
- uses: actions/setup-node@v6
with:
node-version: '20'
cache: pnpm
- run: pnpm install --frozen-lockfile
- run: cp .env.example .env
- run: pnpm test
- run: pnpm build:back
- run: pnpm benchmark:execution -- --runs=10 --output=execution-baseline.json
- uses: actions/upload-artifact@v6
with:
name: execution-baseline
path: execution-baseline.json
+17
View File
@@ -0,0 +1,17 @@
# Assign existing resources to accounts
The panel adds ownership columns automatically while preserving existing data. Unassigned historical data is visible only to administrators. Ordinary users can read their assigned environment variables, task metadata and subscription metadata; dependencies and execution remain administrative operations.
Prefer the authenticated administrator endpoint `PUT /api/user-management/:id/resources` with a body such as `{"resource":"envs","ids":[1,2]}`. Supported resources are `envs`, `crons` and `subscriptions`. Assignment does not run tasks or move files.
The optional local migration helper assigns every unassigned row to one existing account. Stop the panel and back up `data/db/database.sqlite` before using it; set `QL_DIR` or `QL_DATA_DIR` when the data directory differs from the repository default. The helper refuses to create a new empty database.
```bash
node migrate-to-multiuser.js --list-users
node migrate-to-multiuser.js --username=alice --dry-run
node migrate-to-multiuser.js --username=alice
```
Already assigned rows are preserved and writes run in a transaction. Dependencies may receive an owner for bookkeeping, but remain inaccessible to ordinary users. To allocate only selected rows, use the administrator resource endpoint instead of the bulk helper.
See [the permission contract](docs/development/multiuser-permissions.md).
+7
View File
@@ -0,0 +1,7 @@
# Multi-user accounts
Administrators can create, disable and delete accounts in Settings → User Management. New administrator accounts have full access to the shared instance. Ordinary accounts can only read their assigned environment variables, task metadata and subscription metadata. They cannot execute tasks, read shared files or logs, install dependencies, or modify system settings.
Existing data with no owner remains visible only to administrators. The original administrator and its password migration, MFA and sessions are preserved. Accounts from older drafts that used bcrypt require an administrator password reset.
See [the permission contract](docs/development/multiuser-permissions.md) and [resource migration](MIGRATION_GUIDE.md). This draft does not provide isolated execution for untrusted users.
+24
View File
@@ -34,6 +34,7 @@ Timed task management platform supporting Python3, JavaScript, Shell, Typescript
- Support system level notification
- Support dark mode
- Support cell phone operation
- Manage panels through the [remote CLI and Agent Skill](#remote-cli-and-agent-skill)
## Version
@@ -79,6 +80,29 @@ npm i @whyour/qinglong
[View Documentation](https://qinglong.online/guide/user-guide/basic-explanation)
## Remote CLI and Agent Skill
Use the standalone npm package [@whyour/qinglong-cli](https://www.npmjs.com/package/@whyour/qinglong-cli) to manage a remote QingLong panel from your terminal or automation workflows. Create, update and run tasks and subscriptions, inspect logs, and manage applications, environment variables, scripts, configuration, dependencies and other OpenAPI resources. JSON output supports scripts and AI agents.
Requires Node.js >=22.12; Node.js 24 is recommended. Install the package to use `ql`:
```bash
npm install -g @whyour/qinglong-cli
ql login --url https://ql.example.com
ql auth status --json
ql task list --json
ql subscription list --json
ql --help
```
Create an application in the panel's System Settings → Application Settings and grant the required permissions. Enter its Client ID/Secret when prompted during login. Existing tokens are also supported through `QL_URL` and `QL_ACCESS_TOKEN`. Each command supports `--help`; set `QL_LANG=en` for English help.
The npm CLI manages remote panels. The panel's built-in executable is also named `ql`; use a separate installation directory on panel hosts to avoid replacing it. For local script execution and maintenance, see [panel-internal tools](cli/LOCAL.en.md).
[CLI guide](cli/README.en.md) · [简体中文](cli/README.md) · [Full command reference](cli/skills/qinglong-cli/references/openapi.md)
The package includes a [remote management skill](cli/skills/qinglong-cli/SKILL.md) covering authentication, command selection and result verification for AI agents. Local execution and maintenance have a separate [internal skill](cli/skills/qinglong-local/SKILL.md).
## Development
```bash
+24
View File
@@ -36,6 +36,7 @@ Timed task management platform supporting Python3, JavaScript, Shell, Typescript
- 支持系统级通知
- 支持暗黑模式
- 支持手机端操作
- 支持通过 [远程 CLI 与 Agent Skill](#远程-cli-与-agent-skill) 管理面板
## 版本
@@ -81,6 +82,29 @@ npm i @whyour/qinglong
[查看文档](https://qinglong.online/guide/user-guide/basic-explanation)
## 远程 CLI 与 Agent Skill
使用独立 npm 包 [@whyour/qinglong-cli](https://www.npmjs.com/package/@whyour/qinglong-cli),可以从本机终端或自动化流程管理远程青龙面板。支持任务与订阅的创建、修改、运行和日志查看,以及应用、环境变量、脚本、配置、依赖等 OpenAPI 资源;JSON 输出便于脚本和 AI Agent 使用。
需要 Node.js >=22.12,推荐 Node.js 24。安装后使用 `ql` 命令:
```bash
npm install -g @whyour/qinglong-cli
ql login --url https://ql.example.com
ql auth status --json
ql task list --json
ql subscription list --json
ql --help
```
在面板「系统设置 → 应用设置」创建应用并授予所需权限,登录时按提示输入 Client ID/Secret;也支持通过 `QL_URL` 和 `QL_ACCESS_TOKEN` 使用已有令牌。各命令支持 `--help`,`QL_LANG=en` 可切换英文帮助。
npm CLI 用于远程管理。面板内置 `ql` 同样使用这个命令名,已有面板的机器请使用独立安装目录,避免覆盖内置命令。本机脚本执行和运维见 [面板内部工具](cli/LOCAL.md)。
[CLI 中文说明](cli/README.md) · [English guide](cli/README.en.md) · [完整命令参考](cli/skills/qinglong-cli/references/openapi.md)
包内附带 [远程管理 Skill](cli/skills/qinglong-cli/SKILL.md),涵盖认证、命令选择和操作结果检查,可供 AI Agent 使用。本机执行与运维另见 [内部 Skill](cli/skills/qinglong-local/SKILL.md)。
## 开发
```bash
+3 -7
View File
@@ -4,7 +4,6 @@ import { Logger } from 'winston';
import config from '../config';
import * as fs from 'fs/promises';
import { celebrate, Joi } from 'celebrate';
import { join, basename } from 'path';
import { SAMPLE_FILES } from '../config/const';
import { t } from '../shared/i18n';
import ConfigService from '../services/config';
@@ -73,15 +72,12 @@ export default (app: Router) => {
const logger: Logger = Container.get('logger');
try {
const { name, content } = req.body;
// Resolve path first to prevent traversal attacks
let basePath = config.configPath;
if (name.startsWith('data/scripts/')) {
basePath = join(config.rootPath, 'data/scripts');
return res.send({ code: 403, message: t('文件无法访问') });
}
const cleanName = name.replace(/^data\/scripts\//, '');
const normalized = resolveFileAccess(
basePath,
[cleanName],
config.configPath,
[name],
config.blackFileList,
);
if (!normalized) {
+14
View File
@@ -10,12 +10,26 @@ import {
InstanceStatus,
} from '../data/runningInstance';
import { t } from '../shared/i18n';
import cronClient from '../schedule/client';
const route = Router();
export default (app: Router) => {
app.use('/crons', route);
route.use(async (req, res, next) => {
// Keep stop/status callbacks available even when the scheduler is down.
if (['POST', 'PUT', 'DELETE'].includes(req.method) &&
['/', '/run', '/enable', '/disable', '/views/enable', '/views/disable'].includes(req.path)) {
try {
await cronClient.readiness.ensureReady();
} catch (error) {
return next(error);
}
}
return next();
});
route.get(
'/views',
async (req: Request, res: Response, next: NextFunction) => {
+39 -36
View File
@@ -107,42 +107,45 @@ export default (app: Router) => {
},
);
route.get(
'/failures',
async (req: Request, res: Response, next: NextFunction) => {
try {
const rows = (await CrontabStatModel.findAll({
attributes: ['ref_id', [fn('SUM', col('fail_count')), 'fail_count']],
where: { date: dayjs().format('YYYY-MM-DD'), fail_count: { [Op.gt]: 0 } },
group: ['ref_id'],
order: [[fn('SUM', col('fail_count')), 'DESC'], ['ref_id', 'ASC']],
raw: true,
})) as any[];
const crons = rows.length > 0 ? await CrontabModel.findAll({
attributes: ['id', 'name', 'command'],
where: { id: { [Op.in]: rows.map((row) => Number(row.ref_id)) } },
raw: true,
}) : [];
const cronMap = new Map(crons.map((cron) => [cron.id, cron]));
res.send({
code: 200,
data: rows.map((row) => {
const id = Number(row.ref_id);
const cron = cronMap.get(id);
return {
id,
name: cron?.name || cron?.command || tf('任务#%s', id),
command: cron?.command || '',
failCount: Number(row.fail_count),
deleted: !cron,
};
}),
});
} catch (e) {
next(e);
}
},
);
for (const isSuccess of [false, true]) {
route.get(
isSuccess ? '/successes' : '/failures',
async (req: Request, res: Response, next: NextFunction) => {
try {
const countColumn = isSuccess ? 'success_count' : 'fail_count';
const rows = (await CrontabStatModel.findAll({
attributes: ['ref_id', [fn('SUM', col(countColumn)), 'result_count']],
where: { date: dayjs().format('YYYY-MM-DD'), [countColumn]: { [Op.gt]: 0 } },
group: ['ref_id'],
order: [[fn('SUM', col(countColumn)), 'DESC'], ['ref_id', 'ASC']],
raw: true,
})) as any[];
const crons = rows.length > 0 ? await CrontabModel.findAll({
attributes: ['id', 'name', 'command'],
where: { id: { [Op.in]: rows.map((row) => Number(row.ref_id)) } },
raw: true,
}) : [];
const cronMap = new Map(crons.map((cron) => [cron.id, cron]));
res.send({
code: 200,
data: rows.map((row) => {
const id = Number(row.ref_id);
const cron = cronMap.get(id);
return {
id,
name: cron?.name || cron?.command || tf('任务#%s', id),
command: cron?.command || '',
[isSuccess ? 'successCount' : 'failCount']: Number(row.result_count),
deleted: !cron,
};
}),
});
} catch (e) {
next(e);
}
},
);
}
route.get(
'/trend',
+17 -14
View File
@@ -1,24 +1,16 @@
import { Joi, celebrate } from 'celebrate';
import { NextFunction, Request, Response, Router } from 'express';
import fs from 'fs';
import multer from 'multer';
import { Container } from 'typedi';
import { Logger } from 'winston';
import config from '../config';
import { safeJSONParse } from '../config/util';
import { t } from '../shared/i18n';
import EnvService from '../services/env';
const route = Router();
const storage = multer.diskStorage({
destination: function (req, file, cb) {
cb(null, config.scriptPath);
},
filename: function (req, file, cb) {
cb(null, file.originalname);
},
const upload = multer({
storage: multer.memoryStorage(),
limits: { fileSize: 5 * 1024 * 1024, files: 1 },
});
const upload = multer({ storage: storage });
const labelSchema = Joi.array()
.items(Joi.string().trim().required())
.min(1)
@@ -282,12 +274,23 @@ export default (app: Router) => {
const logger: Logger = Container.get('logger');
try {
const envService = Container.get(EnvService);
const fileContent = await fs.promises.readFile(req!.file!.path, 'utf8');
const parseContent = safeJSONParse(fileContent);
let parseContent: unknown;
try {
parseContent = JSON.parse(req.file?.buffer.toString('utf8') || '');
} catch {
return res.status(400).send({ code: 400, message: t('参数错误') });
}
const data = Array.isArray(parseContent)
? parseContent
: [parseContent];
if (data.every((x) => x.name && x.value)) {
if (
data.length > 0 &&
data.every((x) =>
x && typeof x.name === 'string' &&
/^[a-zA-Z_][0-9a-zA-Z_]*$/.test(x.name) &&
typeof x.value === 'string' && x.value.length > 0,
)
) {
const result = await envService.create(
data.map((x) => ({
name: x.name,
+2 -2
View File
@@ -11,8 +11,8 @@ export default (app: Router) => {
try {
const healthService = Container.get(HealthService);
const health = await healthService.check();
res.status(200).send({
code: 200,
res.status(health.status === 'ok' ? 200 : 503).send({
code: health.status === 'ok' ? 200 : 503,
data: health,
});
} catch (err: any) {
+2
View File
@@ -12,6 +12,7 @@ import subscription from './subscription';
import update from './update';
import dashboard from './dashboard';
import health from './health';
import userManagement from './userManagement';
import clientIp from './clientIp';
import retention from './retention';
@@ -30,6 +31,7 @@ export default () => {
update(app);
dashboard(app);
health(app);
userManagement(app);
clientIp(app);
retention(app);
+2
View File
@@ -3,6 +3,7 @@ import { Container } from 'typedi';
import OpenService from '../services/open';
import { Logger } from 'winston';
import { celebrate, Joi } from 'celebrate';
import { rateLimit } from 'express-rate-limit';
const route = Router();
export default (app: Router) => {
@@ -100,6 +101,7 @@ export default (app: Router) => {
route.get(
'/auth/token',
rateLimit({ windowMs: 15 * 60 * 1000, limit: 100 }),
celebrate({
query: {
client_id: Joi.string().required(),
+199 -14
View File
@@ -7,20 +7,62 @@ import { Logger } from 'winston';
import config from '../config';
import * as fs from 'fs/promises';
import { celebrate, Joi } from 'celebrate';
import path, { join, parse } from 'path';
import path, { dirname, join, parse, resolve, sep } from 'path';
import ScriptService from '../services/script';
import ScriptHistoryService from '../services/scriptHistory';
import { t } from '../shared/i18n';
import multer from 'multer';
import { writeFileWithLock } from '../shared/utils';
import {
ScriptHistoryError,
HistoryUnavailableError,
} from '../shared/scriptHistory';
const route = Router();
function isPathAllowed(targetPath: string): boolean {
const resolved = path.resolve(targetPath);
return config.writePathList.some((x) =>
Boolean(resolveFileAccess(x, [resolved], config.blackFileList)),
Boolean(
resolveFileAccess(
x,
[resolved],
// Panel configuration secrets must not restrict user script filenames.
path.resolve(x) === path.resolve(config.configPath)
? config.blackFileList
: [],
),
),
);
}
async function existingWritableRealPath(targetPath: string): Promise<string> {
if (!isPathAllowed(targetPath)) {
throw new ScriptHistoryError(403, '暂无权限');
}
const resolved = resolve(targetPath);
for (const writableRoot of config.writePathList) {
const root = resolve(writableRoot);
const rootPrefix = root.endsWith(sep) ? root : root + sep;
// Keep an explicit normalized boundary check at the filesystem operation.
if (!resolved.startsWith(rootPrefix)) continue;
const realPath = await fs.realpath(resolved);
const realRoot = await fs.realpath(root);
const realRootPrefix = realRoot.endsWith(sep) ? realRoot : realRoot + sep;
if (
!realPath.startsWith(realRootPrefix) ||
!resolveFileAccess(
realRoot,
[realPath],
root === resolve(config.configPath) ? config.blackFileList : [],
)
) {
throw new ScriptHistoryError(403, '暂无权限');
}
return realPath;
}
throw new ScriptHistoryError(403, '暂无权限');
}
const storage = multer.diskStorage({
destination: function (req, file, cb) {
cb(null, config.tmpPath);
@@ -55,11 +97,7 @@ export default (app: Router) => {
];
if (req.query.path) {
if (
!resolveFileAccess(
config.scriptPath,
[req.query.path as string],
config.blackFileList,
)
!resolveFileAccess(config.scriptPath, [req.query.path as string])
) {
return res.send({ code: 403, message: t('暂无权限') });
}
@@ -115,6 +153,82 @@ export default (app: Router) => {
},
);
const historyError = (error: unknown, res: Response, next: NextFunction) => {
if (error instanceof ScriptHistoryError) {
return res.status(error.status).send({
code: error.status,
message: t(error.message),
...(error instanceof HistoryUnavailableError
? { historyUnavailable: true, currentHash: error.currentHash }
: {}),
});
}
return next(error);
};
const historyQuery = {
filename: Joi.string().required(),
path: Joi.string().optional().allow(''),
};
route.get(
'/history',
celebrate({ query: Joi.object(historyQuery).unknown(true) }),
async (req: Request, res: Response, next: NextFunction) => {
try {
const service = Container.get(ScriptHistoryService);
const data = await service.list(
(req.query.path as string) || '',
req.query.filename as string,
);
res.send({ code: 200, data });
} catch (error) {
historyError(error, res, next);
}
},
);
route.get(
'/history/detail',
celebrate({
query: Joi.object({
...historyQuery,
id: Joi.string().uuid().required(),
}).unknown(true),
}),
async (req: Request, res: Response, next: NextFunction) => {
try {
const service = Container.get(ScriptHistoryService);
const data = await service.detail(
(req.query.path as string) || '',
req.query.filename as string,
req.query.id as string,
);
res.send({ code: 200, data });
} catch (error) {
historyError(error, res, next);
}
},
);
route.put(
'/history/restore',
celebrate({
body: Joi.object({
...historyQuery,
id: Joi.string().uuid().required(),
expectedHash: Joi.string().hex().length(64).required(),
}),
}),
async (req: Request, res: Response, next: NextFunction) => {
try {
const { path = '', filename, id, expectedHash } = req.body;
const service = Container.get(ScriptHistoryService);
const data = await service.restore(path, filename, id, expectedHash);
res.send({ code: 200, data });
} catch (error) {
historyError(error, res, next);
}
},
);
route.get('/:file', (req: Request, res: Response) => {
return res.send({
code: 410,
@@ -136,6 +250,11 @@ export default (app: Router) => {
filename: Joi.string().required(),
path: Joi.string().optional().allow(''),
content: Joi.string().optional().allow(''),
skipHistory: Joi.boolean().optional(),
expectedHash: Joi.string()
.hex()
.length(64)
.when('skipHistory', { is: true, then: Joi.required() }),
originFilename: Joi.string().optional().allow(''),
directory: Joi.string().optional().allow(''),
file: Joi.string().optional().allow(''),
@@ -195,8 +314,56 @@ export default (app: Router) => {
if (!isPathAllowed(filePath) || !isPathAllowed(originFilePath)) {
return res.send({ code: 403, message: t('暂无权限') });
}
await fs.mkdir(path, { recursive: true });
// Check the canonical parent directly at the directory creation boundary.
const parentPath = resolve(dirname(filePath));
let parentCreated = false;
for (const writableRoot of config.writePathList) {
const root = resolve(writableRoot);
const rootPrefix = root.endsWith(sep) ? root : root + sep;
if (parentPath === root) {
await fs.mkdir(root, { recursive: true });
} else if (parentPath.startsWith(rootPrefix)) {
await fs.mkdir(parentPath, { recursive: true });
} else {
continue;
}
parentCreated = true;
break;
}
if (!parentCreated) {
return res.send({ code: 403, message: t('暂无权限') });
}
const fileExists = await fileExist(filePath);
if (fileExists && resolveFileAccess(config.scriptPath, [filePath])) {
let removeSource = filename !== originFilename;
let backupSource = '';
if (removeSource) {
const [originRealPath, targetRealPath] = await Promise.all([
existingWritableRealPath(originFilePath),
existingWritableRealPath(filePath),
]);
// Aliases of one script are an in-place save. Removing the source
// would also remove the destination behind a target symlink.
removeSource = originRealPath !== targetRealPath;
backupSource = originRealPath;
}
// Save-as removes the source after committing the destination. Keep
// its original content too: destination history only protects the
// file being overwritten, not the source being deleted.
if (removeSource) {
await fs.copyFile(
backupSource,
join(config.bakPath, originFilename.replace(/\//g, '')),
);
}
const service = Container.get(ScriptHistoryService);
const data = await service.save(path, filename, content, {
skipHistory: req.body.skipHistory,
expectedHash: req.body.expectedHash,
});
if (removeSource) await rmPath(originFilePath);
return res.send({ code: 200, data });
}
if (fileExists) {
await fs.copyFile(
originFilePath,
@@ -209,7 +376,7 @@ export default (app: Router) => {
await writeFileWithLock(filePath, content);
return res.send({ code: 200 });
} catch (e) {
return next(e);
return historyError(e, res, next);
}
},
);
@@ -221,6 +388,11 @@ export default (app: Router) => {
filename: Joi.string().required(),
path: Joi.string().optional().allow(''),
content: Joi.string().required().allow(''),
skipHistory: Joi.boolean().optional(),
expectedHash: Joi.string()
.hex()
.length(64)
.when('skipHistory', { is: true, then: Joi.required() }),
}),
}),
async (req: Request, res: Response, next: NextFunction) => {
@@ -238,10 +410,14 @@ export default (app: Router) => {
message: t('暂无权限'),
});
}
await writeFileWithLock(filePath, content);
return res.send({ code: 200 });
const service = Container.get(ScriptHistoryService);
const data = await service.save(path || '', filename, content, {
skipHistory: req.body.skipHistory,
expectedHash: req.body.expectedHash,
});
return res.send({ code: 200, data });
} catch (e) {
return next(e);
return historyError(e, res, next);
}
},
);
@@ -368,12 +544,21 @@ export default (app: Router) => {
if (!isPathAllowed(filePath)) {
return res.send({ code: 403, message: t('暂无权限') });
}
const logPath = join(config.logPath, path, `${name}.swap`);
const logPath = resolveFileAccess(config.logPath, [
path,
`${name}.swap`,
]);
if (!logPath) {
return res.send({ code: 403, message: t('暂无权限') });
}
const scriptService = Container.get(ScriptService);
const result = await scriptService.stopScript(filePath, pid);
setTimeout(() => {
rmPath(logPath);
const cleanupPath = resolveFileAccess(config.logPath, [logPath]);
if (cleanupPath) {
void rmPath(cleanupPath);
}
}, 3000);
res.send(result);
} catch (e) {
+3 -10
View File
@@ -3,7 +3,7 @@ import { Container } from 'typedi';
import { Logger } from 'winston';
import SubscriptionService from '../services/subscription';
import { celebrate, Joi } from 'celebrate';
import CronExpressionParser from 'cron-parser';
import { isValidCronSchedule } from '../shared/cronSchedule';
const route = Router();
export default (app: Router) => {
@@ -58,10 +58,7 @@ export default (app: Router) => {
async (req: Request, res: Response, next: NextFunction) => {
const logger: Logger = Container.get('logger');
try {
if (
!req.body.schedule ||
CronExpressionParser.parse(req.body.schedule).hasNext()
) {
if (!req.body.schedule || isValidCronSchedule(req.body.schedule)) {
const subscriptionService = Container.get(SubscriptionService);
const data = await subscriptionService.create(req.body);
return res.send({ code: 200, data });
@@ -204,11 +201,7 @@ export default (app: Router) => {
async (req: Request, res: Response, next: NextFunction) => {
const logger: Logger = Container.get('logger');
try {
if (
!req.body.schedule ||
typeof req.body.schedule === 'object' ||
CronExpressionParser.parse(req.body.schedule).hasNext()
) {
if (!req.body.schedule || isValidCronSchedule(req.body.schedule)) {
const subscriptionService = Container.get(SubscriptionService);
const data = await subscriptionService.update(req.body);
return res.send({ code: 200, data });
+18 -10
View File
@@ -8,15 +8,12 @@ import { celebrate, Joi } from 'celebrate';
import UserService from '../services/user';
import { t } from '../shared/i18n';
import { isDefaultAuthInfo } from '../shared/auth';
import {
getUniqPath,
handleLogPath,
parseVersion,
promiseExec,
} from '../config/util';
import { getUniqPath, handleLogPath, parseVersion } from '../config/util';
import { findCronId } from '../shared/cronCommand';
import dayjs from 'dayjs';
import multer from 'multer';
import { logStreamManager } from '../shared/logStreamManager';
import { rateLimit } from 'express-rate-limit';
const route = Router();
const storage = multer.diskStorage({
@@ -249,6 +246,7 @@ export default (app: Router) => {
route.put(
'/command-run',
rateLimit({ windowMs: 60 * 1000, limit: 60 }),
celebrate({
body: Joi.object({
command: Joi.string().required(),
@@ -258,8 +256,13 @@ export default (app: Router) => {
try {
const systemService = Container.get(SystemService);
const command = req.body.command;
const idStr = `cat ${config.crontabFile} | grep -E "${command}" | perl -pe "s|.*ID=(.*) ${command}.*|\\1|" | head -1 | awk -F " " '{print $1}' | xargs echo -n`;
let id = await promiseExec(idStr);
const crontab = await fs
.readFile(config.crontabFile, 'utf8')
.catch((e) => {
if (e.code === 'ENOENT') return '';
throw e;
});
const id = findCronId(crontab, command);
const uniqPath = await getUniqPath(command, id);
const logTime = dayjs().format('YYYY-MM-DD-HH-mm-ss-SSS');
const logPath = `${uniqPath}/${logTime}.log`;
@@ -273,8 +276,13 @@ export default (app: Router) => {
},
onEnd: async (cp, endTime, diff) => {
// Close the stream after task completion
await logStreamManager.closeStream(await handleLogPath(logPath));
res.end();
try {
await logStreamManager.closeStream(
await handleLogPath(logPath),
);
} finally {
res.end();
}
},
onError: async (message: string) => {
res.write(message);
+39 -1
View File
@@ -76,7 +76,17 @@ export default (app: Router) => {
try {
const userService = Container.get(UserService);
const token = getToken(req);
await userService.logout(req.platform, token);
if (req.user?.userId) {
const { default: UserManagementService } = await import(
'../services/userManagement'
);
await Container.get(UserManagementService).revokeToken(
req.user.userId,
token,
);
} else {
await userService.logout(req.platform, token);
}
res.send({ code: 200 });
} catch (e) {
return next(e);
@@ -98,6 +108,16 @@ export default (app: Router) => {
return res.send({ code: 450, message: t('未知错误') });
}
const userService = Container.get(UserService);
if (req.user?.userId) {
const { default: UserManagementService } = await import(
'../services/userManagement'
);
await Container.get(UserManagementService).update({
id: req.user.userId,
...req.body,
});
return res.send({ code: 200 });
}
const result = await userService.updateUsernameAndPassword(req.body);
res.send(result);
} catch (e) {
@@ -110,10 +130,28 @@ export default (app: Router) => {
const logger: Logger = Container.get('logger');
try {
const userService = Container.get(UserService);
if (req.user?.userId) {
const { default: UserManagementService } = await import(
'../services/userManagement'
);
const account = await Container.get(UserManagementService).get(
req.user.userId,
);
return res.send({
code: 200,
data: {
username: account.username,
role: account.role,
id: account.id,
readOnly: account.role === 1,
},
});
}
const authInfo = await userService.getAuthInfo();
res.send({
code: 200,
data: {
role: 0,
username: authInfo.username,
avatar: authInfo.avatar,
twoFactorActivated: authInfo.twoFactorActivated,
+168
View File
@@ -0,0 +1,168 @@
import { Router, Request, Response, NextFunction } from 'express';
import { Container } from 'typedi';
import { celebrate, Joi } from 'celebrate';
import UserManagementService from '../services/userManagement';
import { UserRole, UserStatus } from '../data/user';
import { EnvModel } from '../data/env';
import { CrontabModel } from '../data/cron';
import { SubscriptionModel } from '../data/subscription';
const route = Router();
// Middleware to check if user is admin
const requireAdmin = (req: Request, res: Response, next: NextFunction) => {
if (req.user && req.user.role === UserRole.admin) {
return next();
}
return res.status(403).send({ code: 403, message: '需要管理员权限' });
};
export default (app: Router) => {
app.use('/user-management', route);
// List all users (admin only)
route.get(
'/',
requireAdmin,
async (req: Request, res: Response, next: NextFunction) => {
try {
const userManagementService = Container.get(UserManagementService);
const data = await userManagementService.list(
req.query.searchValue as string,
);
res.send({ code: 200, data });
} catch (e) {
return next(e);
}
},
);
route.put(
'/:id/resources',
requireAdmin,
celebrate({
params: Joi.object({ id: Joi.number().integer().positive().required() }),
body: Joi.object({
resource: Joi.string()
.valid('envs', 'crons', 'subscriptions')
.required(),
ids: Joi.array()
.min(1)
.max(200)
.unique()
.items(Joi.number().integer().positive())
.required(),
}),
}),
async (req, res, next) => {
try {
const id = Number(req.params.id);
await Container.get(UserManagementService).get(id);
const models: Record<string, any> = {
envs: EnvModel,
crons: CrontabModel,
subscriptions: SubscriptionModel,
};
const [count] = await models[req.body.resource].update(
{ userId: id },
{ where: { id: req.body.ids } },
);
return res.json({ code: 200, data: { assigned: count } });
} catch (error) {
return next(error);
}
},
);
// Get a specific user (admin only)
route.get(
'/:id',
requireAdmin,
celebrate({ params: Joi.object({ id: Joi.number().integer().positive().required() }) }),
async (req: Request, res: Response, next: NextFunction) => {
try {
const userManagementService = Container.get(UserManagementService);
const data = await userManagementService.get(Number(req.params.id));
res.send({ code: 200, data });
} catch (e) {
return next(e);
}
},
);
// Create a new user (admin only)
route.post(
'/',
requireAdmin,
celebrate({
body: Joi.object({
username: Joi.string().trim().max(255).required(),
password: Joi.string().min(6).max(1024).required(),
role: Joi.number()
.valid(UserRole.admin, UserRole.user)
.default(UserRole.user),
status: Joi.number()
.valid(UserStatus.active, UserStatus.disabled)
.optional(),
}),
}),
async (req: Request, res: Response, next: NextFunction) => {
try {
const userManagementService = Container.get(UserManagementService);
const data = await userManagementService.create(req.body);
res.send({ code: 200, data, message: '创建用户成功' });
} catch (e: any) {
return res.send({ code: 400, message: e.message });
}
},
);
// Update a user (admin only)
route.put(
'/',
requireAdmin,
celebrate({
body: Joi.object({
id: Joi.number().integer().positive().required(),
username: Joi.string().trim().max(255).required(),
password: Joi.string().min(6).max(1024).optional().allow(''),
role: Joi.number().valid(UserRole.admin, UserRole.user),
status: Joi.number()
.valid(UserStatus.active, UserStatus.disabled)
.optional(),
}),
}),
async (req: Request, res: Response, next: NextFunction) => {
try {
const userManagementService = Container.get(UserManagementService);
const data = await userManagementService.update(req.body);
res.send({ code: 200, data, message: '更新用户成功' });
} catch (e: any) {
return res.send({ code: 400, message: e.message });
}
},
);
// Delete users (admin only)
route.delete(
'/',
requireAdmin,
celebrate({
body: Joi.array()
.min(1)
.max(200)
.unique()
.items(Joi.number().integer().positive())
.required(),
}),
async (req: Request, res: Response, next: NextFunction) => {
try {
const userManagementService = Container.get(UserManagementService);
const count = await userManagementService.delete(req.body);
res.send({ code: 200, data: count, message: '删除用户成功' });
} catch (e) {
return next(e);
}
},
);
};
+49 -40
View File
@@ -1,14 +1,10 @@
import 'reflect-metadata';
import cluster, { type Worker } from 'cluster';
import compression from 'compression';
import cors from 'cors';
import express from 'express';
import helmet from 'helmet';
import type express from 'express';
import { Container } from 'typedi';
import config from './config';
import Logger from './loaders/logger';
import { monitoringMiddleware } from './middlewares/monitoring';
import { errStack } from './config/util';
import { errStack } from './shared/errors';
import { type GrpcServerService } from './services/grpc';
import { type HttpServerService } from './services/http';
@@ -20,24 +16,12 @@ interface WorkerMetadata {
}
class Application {
private app: express.Application;
private httpServerService?: HttpServerService;
private grpcServerService?: GrpcServerService;
private isShuttingDown = false;
private workerMetadataMap = new Map<number, WorkerMetadata>();
private httpWorker?: Worker;
constructor() {
this.app = express();
// 创建一个全局中间件,删除查询参数中的t
this.app.use((req: express.Request, res: express.Response, next: express.NextFunction) => {
if (req.query.t) {
delete req.query.t;
}
next();
});
}
async start() {
try {
if (cluster.isPrimary) {
@@ -79,6 +63,11 @@ class Application {
);
// If gRPC worker died, restart it and wait for it to be ready
if (metadata.serviceType === 'grpc') {
try {
this.httpWorker?.send('scheduler-unavailable');
} catch (error) {
Logger.warn('Unable to notify HTTP worker of scheduler exit');
}
const newGrpcWorker = this.forkWorker('grpc');
this.waitForWorkerReady(newGrpcWorker, 30000)
.then(() => {
@@ -132,7 +121,14 @@ class Application {
}
private forkWorker(serviceType: string): Worker {
const worker = cluster.fork({ SERVICE_TYPE: serviceType });
const workerEnv: NodeJS.ProcessEnv = { SERVICE_TYPE: serviceType };
// PM2's fork launcher is inherited by our own cluster workers. Their APM
// messages go to this primary, not PM2, and duplicate its sampling work.
// Keep primary monitoring and allow restoring the inherited worker APM.
if (process.env.pm_id !== undefined && process.env.QL_WORKER_APM !== 'true') {
workerEnv.pmx = 'false';
}
const worker = cluster.fork(workerEnv);
this.workerMetadataMap.set(worker.id, {
id: worker.id,
@@ -145,17 +141,37 @@ class Application {
}
private async initializeDatabase() {
const dbLoader = await import('./loaders/db');
await dbLoader.default();
const { runStartupProcess } = await import('./shared/startupProcess');
await runStartupProcess(require.resolve('./bootstrap/database'));
}
private setupMiddlewares() {
this.app.use(helmet({
private async setupMiddlewares(): Promise<express.Application> {
// Only the HTTP worker needs an Express application and its middleware.
const [
{ default: createExpress },
{ default: helmet },
{ default: cors },
{ default: compression },
{ monitoringMiddleware },
] = await Promise.all([
import('express'),
import('helmet'),
import('cors'),
import('compression'),
import('./middlewares/monitoring'),
]);
const app = createExpress();
app.use((req, res, next) => {
if (req.query.t) delete req.query.t;
next();
});
app.use(helmet({
contentSecurityPolicy: false,
}));
this.app.use(cors(config.cors));
this.app.use(compression());
this.app.use(monitoringMiddleware);
app.use(cors(config.cors));
app.use(compression());
app.use(monitoringMiddleware);
return app;
}
private setupMasterShutdown() {
@@ -234,16 +250,16 @@ class Application {
const { initGrpcCerts } = await import('./config/grpcCerts');
await initGrpcCerts();
this.setupMiddlewares();
const app = await this.setupMiddlewares();
const { HttpServerService } = await import('./services/http');
this.httpServerService = Container.get(HttpServerService);
const appLoader = await import('./loaders/app');
await appLoader.default({ app: this.app });
await appLoader.default({ app });
const server = await this.httpServerService.initialize(
this.app,
app,
config.port,
);
@@ -264,17 +280,10 @@ class Application {
process.on('message', async (msg) => {
if (msg === 'shutdown') {
this.gracefulShutdown(serviceType);
} else if (msg === 'reregister-crons' && serviceType === 'http') {
// Re-register cron jobs when gRPC worker restarts
try {
Logger.info('[boot] Received reregister-crons message, re-registering cron jobs...');
const CronService = (await import('./services/cron')).default;
const cronService = Container.get(CronService);
await cronService.autosave_crontab();
Logger.info('[boot] Cron jobs re-registered successfully');
} catch (error) {
Logger.error(`[boot] Failed to re-register cron jobs:\n${errStack(error)}`);
}
} else if (serviceType === 'http' &&
(msg === 'reregister-crons' || msg === 'scheduler-unavailable')) {
const { default: cronClient } = await import('./schedule/client');
cronClient.readiness.invalidate();
}
});
+26
View File
@@ -0,0 +1,26 @@
import { errStack } from '../shared/errors';
// Also terminate if the primary disappears without forwarding a signal.
const onParentDisconnect = () => process.exit(1);
process.once('disconnect', onParentDisconnect);
async function initializeDatabase() {
const { sequelize } = await import('../data');
try {
const { default: loadDatabase } = await import('../loaders/db');
await loadDatabase();
} finally {
await sequelize.close();
}
}
initializeDatabase()
.catch((error) => {
console.error(`[boot] Database initialization failed:\n${errStack(error)}`);
process.exitCode = 1;
})
.finally(() => {
process.removeListener('disconnect', onParentDisconnect);
// Let pending log writes drain and exit naturally after closing SQLite.
if (process.connected) process.disconnect();
});
+1
View File
@@ -51,6 +51,7 @@ export const NotificationModeStringMap = {
19: 'ntfy',
20: 'wxPusherBot',
21: 'wxPusherSpt',
22: 'wpush',
} as const;
export const LINUX_DEPENDENCE_COMMAND: Record<
+93 -45
View File
@@ -1,6 +1,5 @@
import { execSync } from 'child_process';
import { execFileSync } from 'child_process';
import * as fs from 'fs/promises';
import * as os from 'os';
import path from 'path';
import config from './index';
import { fileExist } from './util';
@@ -24,71 +23,120 @@ const clientCertPath = path.join(certDir, 'client.crt');
let cachedConfig: GrpcTlsConfig | null = null;
function run(cmd: string, execOpts?: Record<string, unknown>): string {
const opts = { stdio: 'pipe', timeout: 30000, encoding: 'utf-8', ...execOpts } as any;
return (execSync(cmd, opts) as string).trim();
}
async function tmpFile(prefix: string): Promise<string> {
const dir = (await fileExist(certDir)) ? certDir : os.tmpdir();
await fs.mkdir(dir, { recursive: true });
return path.join(dir, `.${prefix}_${Date.now()}_${Math.random().toString(36).slice(2)}.pem`);
function run(args: string[]): string {
return execFileSync('openssl', args, {
stdio: 'pipe',
timeout: 30000,
encoding: 'utf-8',
}).trim();
}
async function generateAllCerts(): Promise<GrpcTlsConfig> {
Logger.info('[boot] Generating gRPC mTLS certificates...');
const caKeyTmp = await tmpFile('ca_key');
const caCertTmp = await tmpFile('ca_cert');
const serverKeyTmp = await tmpFile('server_key');
const serverCsrTmp = await tmpFile('server_csr');
const serverExtTmp = await tmpFile('server_ext');
const clientKeyTmp = await tmpFile('client_key');
const clientCsrTmp = await tmpFile('client_csr');
const clientExtTmp = await tmpFile('client_ext');
const srlTmp = path.join(path.dirname(caKeyTmp), '.grpc_ca.srl');
await fs.mkdir(certDir, { recursive: true, mode: 0o700 });
// Atomically reserve a private workspace for keys, CSRs and OpenSSL side files.
const workspace = await fs.mkdtemp(path.join(certDir, '.generate-'));
const caKeyTmp = path.join(workspace, 'ca.key');
const caCertTmp = path.join(workspace, 'ca.crt');
const serverKeyTmp = path.join(workspace, 'server.key');
const serverCsrTmp = path.join(workspace, 'server.csr');
const serverExtTmp = path.join(workspace, 'server.ext');
const clientKeyTmp = path.join(workspace, 'client.key');
const clientCsrTmp = path.join(workspace, 'client.csr');
const clientExtTmp = path.join(workspace, 'client.ext');
const cleanup = async () => {
for (const f of [caKeyTmp, caCertTmp, serverKeyTmp, serverCsrTmp, serverExtTmp,
clientKeyTmp, clientCsrTmp, clientExtTmp, srlTmp]) {
try { await fs.unlink(f); } catch {}
}
await fs.rm(workspace, { recursive: true, force: true });
};
try {
// 1. CA(私钥直接存盘,证书写入临时文件供签发使用)
run(`openssl genrsa -out '${caKeyTmp}' 2048 2>/dev/null`);
run(`openssl req -new -x509 -days 3650 -key '${caKeyTmp}' -out '${caCertTmp}' -subj '/CN=qinglong-ca/O=qinglong/C=CN' 2>/dev/null`);
run(['genrsa', '-out', caKeyTmp, '2048']);
run([
'req',
'-new',
'-x509',
'-days',
'3650',
'-key',
caKeyTmp,
'-out',
caCertTmp,
'-subj',
'/CN=qinglong-ca/O=qinglong/C=CN',
]);
const caKey = await fs.readFile(caKeyTmp, 'utf-8');
const caCert = await fs.readFile(caCertTmp, 'utf-8');
await fs.mkdir(certDir, { recursive: true });
await fs.writeFile(caKeyPath, caKey, { mode: 0o600 });
// 2. 服务端
run(`openssl genrsa -out '${serverKeyTmp}' 2048 2>/dev/null`);
run(`openssl req -new -key '${serverKeyTmp}' -out '${serverCsrTmp}' -subj '/CN=grpc-server' 2>/dev/null`);
await fs.writeFile(serverExtTmp, 'subjectAltName=DNS:localhost,IP:127.0.0.1,IP:::1\n');
const serverCert = run(
`openssl x509 -req -days 3650 -in '${serverCsrTmp}' -CA '${caCertTmp}' -CAkey '${caKeyTmp}' -CAcreateserial -extfile '${serverExtTmp}' 2>/dev/null`,
run(['genrsa', '-out', serverKeyTmp, '2048']);
run([
'req',
'-new',
'-key',
serverKeyTmp,
'-out',
serverCsrTmp,
'-subj',
'/CN=grpc-server',
]);
await fs.writeFile(
serverExtTmp,
'subjectAltName=DNS:localhost,IP:127.0.0.1,IP:::1\n',
);
const serverCert = run([
'x509',
'-req',
'-days',
'3650',
'-in',
serverCsrTmp,
'-CA',
caCertTmp,
'-CAkey',
caKeyTmp,
'-CAserial',
path.join(workspace, 'ca.srl'),
'-CAcreateserial',
'-extfile',
serverExtTmp,
]);
const serverKey = await fs.readFile(serverKeyTmp, 'utf-8');
// 3. 客户端
run(`openssl genrsa -out '${clientKeyTmp}' 2048 2>/dev/null`);
run(`openssl req -new -key '${clientKeyTmp}' -out '${clientCsrTmp}' -subj '/CN=grpc-client' 2>/dev/null`);
run(['genrsa', '-out', clientKeyTmp, '2048']);
run([
'req',
'-new',
'-key',
clientKeyTmp,
'-out',
clientCsrTmp,
'-subj',
'/CN=grpc-client',
]);
await fs.writeFile(clientExtTmp, 'extendedKeyUsage=clientAuth\n');
const clientCert = run(
`openssl x509 -req -days 3650 -in '${clientCsrTmp}' -CA '${caCertTmp}' -CAkey '${caKeyTmp}' -CAcreateserial -extfile '${clientExtTmp}' 2>/dev/null`,
);
const clientCert = run([
'x509',
'-req',
'-days',
'3650',
'-in',
clientCsrTmp,
'-CA',
caCertTmp,
'-CAkey',
caKeyTmp,
'-CAserial',
path.join(workspace, 'ca.srl'),
'-CAcreateserial',
'-extfile',
clientExtTmp,
]);
const clientKey = await fs.readFile(clientKeyTmp, 'utf-8');
await cleanup();
Logger.info('[boot] gRPC mTLS certificates generated successfully');
return { caCert, serverCert, serverKey, clientCert, clientKey };
} catch (e) {
} finally {
await cleanup();
throw e;
}
}
+23 -26
View File
@@ -1,5 +1,7 @@
import { Subscription } from '../data/subscription';
import isNil from 'lodash/isNil';
import { shellQuote } from '../shared/shellQuote';
import { getSubscriptionSshAlias } from '../shared/subscriptionPath';
export function formatUrl(doc: Subscription) {
let url = doc.url;
@@ -7,7 +9,7 @@ export function formatUrl(doc: Subscription) {
if (doc.type === 'private-repo') {
if (doc.pull_type === 'ssh-key') {
host = doc.url!.replace(/.*\@([^\:]+)\:.*/, '$1');
url = doc.url!.replace(host, doc.alias);
url = doc.url!.replace(host, getSubscriptionSshAlias(doc.alias));
} else {
host = doc.url!.replace(/.*\:\/\/([^\/]+)\/.*/, '$1');
const { username, password } = doc.pull_option as any;
@@ -18,29 +20,24 @@ export function formatUrl(doc: Subscription) {
}
export function formatCommand(doc: Subscription, url?: string) {
let command = `SUB_ID=${doc.id} ql `;
let _url = url || formatUrl(doc).url;
const {
type,
whitelist,
blacklist,
dependences,
branch,
extensions,
proxy,
autoAddCron,
autoDelCron,
} = doc;
if (type === 'file') {
command += `raw "${_url}" "${proxy || ''}" "${
isNil(autoAddCron) ? true : Boolean(autoAddCron)
}" "${isNil(autoDelCron) ? true : Boolean(autoDelCron)}"`;
} else {
command += `repo "${_url}" "${whitelist || ''}" "${blacklist || ''}" "${
dependences || ''
}" "${branch || ''}" "${extensions || ''}" "${proxy || ''}" "${
isNil(autoAddCron) ? true : Boolean(autoAddCron)
}" "${isNil(autoDelCron) ? true : Boolean(autoDelCron)}"`;
}
return command;
const args =
doc.type === 'file'
? ['raw', url || formatUrl(doc).url, doc.proxy || '']
: [
'repo',
url || formatUrl(doc).url,
doc.whitelist || '',
doc.blacklist || '',
doc.dependences || '',
doc.branch || '',
doc.extensions || '',
doc.proxy || '',
];
args.push(
String(isNil(doc.autoAddCron) ? true : Boolean(doc.autoAddCron)),
String(isNil(doc.autoDelCron) ? true : Boolean(doc.autoDelCron)),
);
return `SUB_ID=${shellQuote(String(doc.id))} ql ${args
.map((arg) => shellQuote(arg!))
.join(' ')}`;
}
+71 -20
View File
@@ -12,6 +12,7 @@ import { DependenceTypes } from '../data/dependence';
import { FormData } from 'undici';
import os from 'os';
import { maybeSudo, isInContainer } from './container';
import { resolveFileAccess } from '../shared/fileAccess';
export * from './share';
@@ -144,7 +145,8 @@ export async function handleLogPath(
logPath: string,
data: string = '',
): Promise<string> {
const absolutePath = path.resolve(config.logPath, logPath);
const absolutePath = resolveFileAccess(config.logPath, [logPath]);
if (!absolutePath) throw new Error('Log path is outside the log directory');
const logFileExist = await fileExist(absolutePath);
if (!logFileExist) {
await createFile(absolutePath, data);
@@ -487,18 +489,69 @@ export function psTree(pid: number): Promise<number[]> {
});
}
export async function killTask(pid: number) {
const pids = await psTree(pid);
if (pids.length) {
export async function killTask(pid: number, waitForExit = false) {
const descendants = await psTree(pid);
const signal = (target: number, sig: NodeJS.Signals) => {
try {
[pid, ...pids].reverse().forEach((x) => {
process.kill(x, 15);
});
} catch (error) { }
} else {
process.kill(pid, 2);
process.kill(target, sig);
} catch (error: any) {
if (error.code !== 'ESRCH') throw error;
}
};
if (!waitForExit) {
if (descendants.length) {
// A child may exit after psTree; keep signalling the remaining tree.
for (const target of [pid, ...descendants].reverse()) {
signal(target, 'SIGTERM');
}
} else signal(pid, 'SIGINT');
return;
}
const pids = [...descendants.reverse(), pid];
for (const target of pids) signal(target, 'SIGTERM');
const alive = async (target: number) => {
try {
process.kill(target, 0);
} catch (error: any) {
if (error.code === 'ESRCH') return false;
throw error;
}
if (process.platform === 'linux') {
try {
const stat = await fs.readFile(`/proc/${target}/stat`, 'utf8');
// The command field may contain spaces and parentheses. Zombies have
// exited even while their parent has not reaped the PID yet.
const state = stat.slice(stat.lastIndexOf(')') + 2).split(' ')[0];
if (['Z', 'X', 'x'].includes(state)) return false;
} catch {
// /proc may be unavailable, or the process may have just exited.
// Retain the portable signal probe rather than assuming it is dead.
try {
process.kill(target, 0);
} catch (error: any) {
if (error.code === 'ESRCH') return false;
throw error;
}
}
}
return true;
};
const wait = async () => {
const deadline = Date.now() + 1000;
let remaining = pids;
while (true) {
const states = await Promise.all(remaining.map(alive));
remaining = remaining.filter((_, index) => states[index]);
if (!remaining.length || Date.now() >= deadline) return remaining;
await new Promise((resolve) => setTimeout(resolve, 20));
}
};
let remaining = await wait();
if (!remaining.length) return;
for (const target of remaining) signal(target, 'SIGKILL');
remaining = await wait();
if (remaining.length)
throw new Error(`Task processes did not exit: ${remaining.join(', ')}`);
}
export async function getPid(cmd: string) {
@@ -598,11 +651,7 @@ export function safeJSONParse(value?: string) {
}
}
export function errStack(error: unknown): string {
return error instanceof Error && error.stack
? error.stack
: String(error);
}
export { errStack } from '../shared/errors';
export async function rmPath(path: string) {
try {
@@ -797,23 +846,25 @@ async function _updateLinuxMirror(
mirrorDomainWithScheme: string,
): Promise<string> {
const S = isInContainer() ? 'sudo ' : '';
// APT otherwise returns zero when index downloads fail with transient errors.
const aptUpdate = `${S}apt-get update --error-on=any`;
let filePath: string, currentDomainWithScheme: string | null;
switch (osType) {
case 'Debian':
filePath = '/etc/apt/sources.list.d/debian.sources';
currentDomainWithScheme = await getCurrentMirrorDomain(filePath);
if (currentDomainWithScheme) {
return `${S}sed -i 's|${currentDomainWithScheme}|${mirrorDomainWithScheme || 'http://deb.debian.org'}|g' ${filePath} || (${S}mkdir -p /etc/apt/sources.list.d && echo -e "Types: deb\\nURIs: ${mirrorDomainWithScheme || 'http://deb.debian.org'}\\nSuites: \\$(grep VERSION_CODENAME /etc/os-release | cut -d= -f2) \\$(grep VERSION_CODENAME /etc/os-release | cut -d= -f2)-updates\\nComponents: main\\nSigned-By: /usr/share/keyrings/debian-archive-keyring.gpg" | ${S}tee ${filePath}) && ${S}apt-get update`;
return `${S}sed -i 's|${currentDomainWithScheme}|${mirrorDomainWithScheme || 'http://deb.debian.org'}|g' ${filePath} || (${S}mkdir -p /etc/apt/sources.list.d && echo -e "Types: deb\\nURIs: ${mirrorDomainWithScheme || 'http://deb.debian.org'}\\nSuites: \\$(grep VERSION_CODENAME /etc/os-release | cut -d= -f2) \\$(grep VERSION_CODENAME /etc/os-release | cut -d= -f2)-updates\\nComponents: main\\nSigned-By: /usr/share/keyrings/debian-archive-keyring.gpg" | ${S}tee ${filePath}) && ${aptUpdate}`;
} else {
return `${S}mkdir -p /etc/apt/sources.list.d && echo -e "Types: deb\\nURIs: ${mirrorDomainWithScheme || 'http://deb.debian.org'}\\nSuites: \\$(grep VERSION_CODENAME /etc/os-release | cut -d= -f2) \\$(grep VERSION_CODENAME /etc/os-release | cut -d= -f2)-updates\\nComponents: main\\nSigned-By: /usr/share/keyrings/debian-archive-keyring.gpg" | ${S}tee ${filePath} && ${S}apt-get update`;
return `${S}mkdir -p /etc/apt/sources.list.d && echo -e "Types: deb\\nURIs: ${mirrorDomainWithScheme || 'http://deb.debian.org'}\\nSuites: \\$(grep VERSION_CODENAME /etc/os-release | cut -d= -f2) \\$(grep VERSION_CODENAME /etc/os-release | cut -d= -f2)-updates\\nComponents: main\\nSigned-By: /usr/share/keyrings/debian-archive-keyring.gpg" | ${S}tee ${filePath} && ${aptUpdate}`;
}
case 'Ubuntu':
filePath = '/etc/apt/sources.list.d/ubuntu.sources';
currentDomainWithScheme = await getCurrentMirrorDomain(filePath);
if (currentDomainWithScheme) {
return `${S}sed -i 's|${currentDomainWithScheme}|${mirrorDomainWithScheme || 'http://archive.ubuntu.com'}|g' ${filePath} || (${S}mkdir -p /etc/apt/sources.list.d && echo -e "Types: deb\\nURIs: ${mirrorDomainWithScheme || 'http://archive.ubuntu.com'}\\nSuites: \\$(grep VERSION_CODENAME /etc/os-release | cut -d= -f2) \\$(grep VERSION_CODENAME /etc/os-release | cut -d= -f2)-updates \\$(grep VERSION_CODENAME /etc/os-release | cut -d= -f2)-backports\\nComponents: main restricted universe multiverse\\nSigned-By: /usr/share/keyrings/ubuntu-archive-keyring.gpg" | ${S}tee ${filePath}) && ${S}apt-get update`;
return `${S}sed -i 's|${currentDomainWithScheme}|${mirrorDomainWithScheme || 'http://archive.ubuntu.com'}|g' ${filePath} || (${S}mkdir -p /etc/apt/sources.list.d && echo -e "Types: deb\\nURIs: ${mirrorDomainWithScheme || 'http://archive.ubuntu.com'}\\nSuites: \\$(grep VERSION_CODENAME /etc/os-release | cut -d= -f2) \\$(grep VERSION_CODENAME /etc/os-release | cut -d= -f2)-updates \\$(grep VERSION_CODENAME /etc/os-release | cut -d= -f2)-backports\\nComponents: main restricted universe multiverse\\nSigned-By: /usr/share/keyrings/ubuntu-archive-keyring.gpg" | ${S}tee ${filePath}) && ${aptUpdate}`;
} else {
return `${S}mkdir -p /etc/apt/sources.list.d && echo -e "Types: deb\\nURIs: ${mirrorDomainWithScheme || 'http://archive.ubuntu.com'}\\nSuites: \\$(grep VERSION_CODENAME /etc/os-release | cut -d= -f2) \\$(grep VERSION_CODENAME /etc/os-release | cut -d= -f2)-updates \\$(grep VERSION_CODENAME /etc/os-release | cut -d= -f2)-backports\\nComponents: main restricted universe multiverse\\nSigned-By: /usr/share/keyrings/ubuntu-archive-keyring.gpg" | ${S}tee ${filePath} && ${S}apt-get update`;
return `${S}mkdir -p /etc/apt/sources.list.d && echo -e "Types: deb\\nURIs: ${mirrorDomainWithScheme || 'http://archive.ubuntu.com'}\\nSuites: \\$(grep VERSION_CODENAME /etc/os-release | cut -d= -f2) \\$(grep VERSION_CODENAME /etc/os-release | cut -d= -f2)-updates \\$(grep VERSION_CODENAME /etc/os-release | cut -d= -f2)-backports\\nComponents: main restricted universe multiverse\\nSigned-By: /usr/share/keyrings/ubuntu-archive-keyring.gpg" | ${S}tee ${filePath} && ${aptUpdate}`;
}
case 'Alpine':
filePath = '/etc/apk/repositories';
+5
View File
@@ -13,6 +13,7 @@ export class Crontab {
pid?: number;
isDisabled?: 1 | 0;
log_path?: string;
queued_token?: string | null;
isPinned?: 1 | 0;
labels?: string[];
last_running_time?: number;
@@ -24,6 +25,7 @@ export class Crontab {
log_name?: string;
allow_multiple_instances?: 1 | 0;
work_dir?: string;
userId?: number;
constructor(options: Crontab) {
this.name = options.name;
@@ -51,6 +53,7 @@ export class Crontab {
this.log_name = options.log_name;
this.allow_multiple_instances = options.allow_multiple_instances || 0;
this.work_dir = options.work_dir;
this.userId = options.userId;
}
}
@@ -83,6 +86,7 @@ export const CrontabModel = sequelize.define<CronInstance>('Crontab', {
isDisabled: DataTypes.NUMBER,
isPinned: DataTypes.NUMBER,
log_path: DataTypes.STRING,
queued_token: DataTypes.STRING,
labels: DataTypes.JSON,
last_running_time: DataTypes.NUMBER,
last_execution_time: DataTypes.NUMBER,
@@ -93,4 +97,5 @@ export const CrontabModel = sequelize.define<CronInstance>('Crontab', {
log_name: DataTypes.STRING,
allow_multiple_instances: DataTypes.NUMBER,
work_dir: DataTypes.STRING,
userId: { type: DataTypes.INTEGER, allowNull: true },
});
+3
View File
@@ -9,6 +9,7 @@ export class Dependence {
name: string;
log?: string[];
remark?: string;
userId?: number;
constructor(options: Dependence) {
this.id = options.id;
@@ -21,6 +22,7 @@ export class Dependence {
this.name = options.name.trim();
this.log = options.log || [];
this.remark = options.remark || '';
this.userId = options.userId;
}
}
@@ -59,5 +61,6 @@ export const DependenceModel = sequelize.define<DependenceInstance>(
status: DataTypes.NUMBER,
log: DataTypes.JSON,
remark: DataTypes.STRING,
userId: { type: DataTypes.NUMBER, allowNull: true },
},
);
+3
View File
@@ -9,6 +9,7 @@ export class Env {
position?: number;
name?: string;
remarks?: string;
userId?: number;
isPinned?: 1 | 0;
labels?: string[];
@@ -23,6 +24,7 @@ export class Env {
this.position = options.position;
this.name = options.name;
this.remarks = options.remarks || '';
this.userId = options.userId;
this.isPinned = options.isPinned || 0;
this.labels = options.labels || [];
}
@@ -46,6 +48,7 @@ export const EnvModel = sequelize.define<EnvInstance>('Env', {
position: DataTypes.NUMBER,
name: { type: DataTypes.STRING, unique: 'compositeIndex' },
remarks: DataTypes.STRING,
userId: { type: DataTypes.NUMBER, allowNull: true },
isPinned: DataTypes.NUMBER,
labels: DataTypes.JSON,
});
+9 -1
View File
@@ -21,6 +21,7 @@ export enum NotificationMode {
'ntfy' = 'ntfy',
'wxPusherBot' = 'wxPusherBot',
'openiLink' = 'openiLink',
'wpush' = 'wpush',
}
abstract class NotificationBaseInfo {
@@ -172,6 +173,12 @@ export class OpeniLinkNotification extends NotificationBaseInfo {
public openiLinkContextToken = '';
}
export class WpushNotification extends NotificationBaseInfo {
public wpushApiKey = '';
public wpushChannel = '';
public wpushTopicCode = '';
}
export interface NotificationInfo
extends GoCqHttpBotNotification,
GotifyNotification,
@@ -195,4 +202,5 @@ export interface NotificationInfo
NtfyNotification,
WxPusherBotNotification,
WxPusherSptNotification,
OpeniLinkNotification {}
OpeniLinkNotification,
WpushNotification {}
+4 -1
View File
@@ -20,7 +20,7 @@ export class Subscription {
pull_option?:
| { private_key: string }
| { username: string; password: string };
pid?: number;
pid?: number | null;
is_disabled?: 1 | 0;
log_path?: string;
alias: string;
@@ -31,6 +31,7 @@ export class Subscription {
proxy?: string;
autoAddCron?: 1 | 0;
autoDelCron?: 1 | 0;
userId?: number;
constructor(options: Subscription) {
this.id = options.id;
@@ -60,6 +61,7 @@ export class Subscription {
this.proxy = options.proxy;
this.autoAddCron = options.autoAddCron ? 1 : 0;
this.autoDelCron = options.autoDelCron ? 1 : 0;
this.userId = options.userId;
}
}
@@ -111,5 +113,6 @@ export const SubscriptionModel = sequelize.define<SubscriptionInstance>(
proxy: { type: DataTypes.STRING, allowNull: true },
autoAddCron: { type: DataTypes.NUMBER, allowNull: true },
autoDelCron: { type: DataTypes.NUMBER, allowNull: true },
userId: { type: DataTypes.NUMBER, allowNull: true },
},
);
+59
View File
@@ -0,0 +1,59 @@
import { sequelize } from '.';
import { DataTypes, Model } from 'sequelize';
export class User {
id?: number;
username: string;
password: string;
role: UserRole;
status: UserStatus;
createdAt?: Date;
updatedAt?: Date;
sessions?: { digest: string; platform: string }[];
constructor(options: User) {
this.id = options.id;
this.username = options.username;
this.password = options.password;
this.role = options.role ?? UserRole.user;
this.status =
typeof options.status === 'number' && UserStatus[options.status]
? options.status
: UserStatus.active;
this.createdAt = options.createdAt;
this.updatedAt = options.updatedAt;
this.sessions = options.sessions || [];
}
}
export enum UserRole {
'admin' = 0,
'user' = 1,
}
export enum UserStatus {
'active' = 0,
'disabled' = 1,
}
export interface UserInstance extends Model<User, User>, User {}
export const UserModel = sequelize.define<UserInstance>('User', {
sessions: { type: DataTypes.JSON, allowNull: false, defaultValue: [] },
username: {
type: DataTypes.STRING,
unique: true,
allowNull: false,
},
password: {
type: DataTypes.STRING,
allowNull: false,
},
role: {
type: DataTypes.INTEGER,
defaultValue: UserRole.user,
},
status: {
type: DataTypes.INTEGER,
defaultValue: UserStatus.active,
},
});
+7 -35
View File
@@ -9,6 +9,9 @@ import { CrontabViewModel } from '../data/cronView';
import { CrontabStatModel } from '../data/cronStats';
import { RunningInstanceModel } from '../data/runningInstance';
import { sequelize } from '../data';
import { UserModel } from '../data/user';
import { migrateUserSchema } from '../shared/userSchemaMigrations';
import { migrateSchema } from '../shared/schemaMigrations';
export default async () => {
try {
@@ -22,44 +25,13 @@ export default async () => {
await CrontabStatModel.sync();
await RunningInstanceModel.sync();
// 初始化新增字段
const migrations = [
{
table: 'CrontabViews',
column: 'filterRelation',
type: 'VARCHAR(255)',
},
{ table: 'Subscriptions', column: 'proxy', type: 'VARCHAR(255)' },
{ table: 'CrontabViews', column: 'type', type: 'NUMBER' },
{ table: 'Subscriptions', column: 'autoAddCron', type: 'NUMBER' },
{ table: 'Subscriptions', column: 'autoDelCron', type: 'NUMBER' },
{ table: 'Crontabs', column: 'sub_id', type: 'NUMBER' },
{ table: 'Crontabs', column: 'extra_schedules', type: 'JSON' },
{ table: 'Crontabs', column: 'task_before', type: 'TEXT' },
{ table: 'Crontabs', column: 'task_after', type: 'TEXT' },
{ table: 'Crontabs', column: 'log_name', type: 'VARCHAR(255)' },
{
table: 'Crontabs',
column: 'allow_multiple_instances',
type: 'NUMBER',
},
{ table: 'Crontabs', column: 'work_dir', type: 'VARCHAR(255)' },
{ table: 'Envs', column: 'isPinned', type: 'NUMBER' },
{ table: 'Envs', column: 'labels', type: 'JSON' },
];
for (const migration of migrations) {
try {
await sequelize.query(
`alter table ${migration.table} add column ${migration.column} ${migration.type}`,
);
} catch (error) {
// Column already exists or other error, continue
}
}
await UserModel.sync();
await migrateSchema(sequelize);
await migrateUserSchema(sequelize);
Logger.info('[boot] DB loaded');
} catch (error) {
Logger.error('[boot] DB load failed', error);
throw error;
}
};
+45 -1
View File
@@ -17,6 +17,50 @@ async function linkCommand() {
}
async function linkCommandToDir(commandDir: string) {
const cliRoot = process.env.QL_CLI_ROOT;
if (cliRoot) {
if (!path.isAbsolute(cliRoot)) {
throw new Error('QL_CLI_ROOT must be an absolute CLI installation path');
}
const { installCliEntrypoints } = require(path.join(
cliRoot,
'dist/local/entrypoints.js',
));
await installCliEntrypoints(commandDir, cliRoot);
const { installCronEntrypoint } = require(path.join(
cliRoot,
'dist/local/cronEntrypoint.js',
));
await installCronEntrypoint(commandDir, cliRoot, config.crontabFile, {
...process.env,
QL_DIR: config.rootPath,
QL_DATA_DIR: config.dataPath,
});
// Container-wide legacy links may precede ~/bin. Ensure panel children use
// the explicitly selected entries without changing system-wide links.
process.env.PATH = [
commandDir,
...(process.env.PATH?.split(path.delimiter) ?? []).filter(
(entry) => entry !== commandDir,
),
].join(path.delimiter);
return;
}
const cronBridge = path.join(commandDir, 'crontab');
try {
if (
(await fs.lstat(cronBridge)).isFile() &&
(await fs.readFile(cronBridge, 'utf8')).includes(
'// QingLong CLI crontab bridge',
)
) {
await fs.unlink(cronBridge);
}
} catch (error) {
if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error;
}
const linkShell = [
{
src: 'update.sh',
@@ -39,7 +83,7 @@ async function linkCommandToDir(commandDir: string) {
if (stats) {
await fs.unlink(tmpTarget);
}
} catch (error) { }
} catch (error) {}
await fs.symlink(source, tmpTarget);
await fs.rename(tmpTarget, target);
+33 -1
View File
@@ -13,7 +13,10 @@ import { isValidToken, isDefaultAuthInfo } from '../shared/auth';
import { AuthInfo } from '../data/system';
import path from 'path';
import { t } from '../shared/i18n';
import { Container } from 'typedi';
import jwt from 'jsonwebtoken';
import { AppScope } from '../data/open';
import uniqueConstraintError from '../middlewares/uniqueConstraintError';
import protectedPathCase from '../middlewares/protectedPathCase';
function resolveTrustProxy(value = process.env.QL_TRUST_PROXY) {
@@ -143,6 +146,26 @@ export default ({ app }: { app: Application }) => {
const authInfo = await shareStore.getAuthInfo();
if (isValidToken(authInfo, headerToken, req.platform, config.jwt.secret)) {
req.user = { role: 0 };
return next();
}
const claims = jwt.decode(headerToken);
let account = null;
if (
typeof claims === 'object' &&
Number.isSafeInteger(claims?.userId) &&
claims!.userId > 0
) {
const { default: UserManagementService } = await import(
'../services/userManagement'
);
account = await Container.get(UserManagementService).validateToken(
headerToken,
req.platform,
);
}
if (account) {
req.user = { userId: account.id, role: account.role };
return next();
}
@@ -176,7 +199,15 @@ export default ({ app }: { app: Application }) => {
});
app.use(rewrite('/open/*', '/api/$1'));
app.use(config.api.prefix, routes());
app.use(
config.api.prefix,
async (req, res, next) => {
if (!req.user || req.user.role === 0) return next();
const { userAccess } = await import('../middlewares/userAccess');
return userAccess(req, res, next);
},
routes(),
);
app.get('*', (_, res, next) => {
const indexPath = path.join(frontendPath, 'index.html');
@@ -190,6 +221,7 @@ export default ({ app }: { app: Application }) => {
});
app.use(errors());
app.use(uniqueConstraintError);
app.use(
(
+14 -26
View File
@@ -13,11 +13,11 @@ import { AuthDataType, SystemModel } from '../data/system';
import SystemService from '../services/system';
import UserService from '../services/user';
import { writeFile, readFile } from 'fs/promises';
import { createRandomString, fileExist, isDemoEnv, safeJSONParse } from '../config/util';
import { fileExist, isDemoEnv, safeJSONParse } from '../config/util';
import OpenService from '../services/open';
import { shareStore } from '../shared/store';
import Logger from './logger';
import { AppModel } from '../data/open';
import cronClient from '../schedule/client';
import { InstanceStatus, RunningInstanceModel } from '../data/runningInstance';
import { setLang, systemLang } from '../shared/i18n';
@@ -30,24 +30,7 @@ export default async () => {
const openService = Container.get(OpenService);
// 初始化增加系统配置
let systemApp = (
await AppModel.findOne({
where: { name: 'system' },
})
)?.get({ plain: true });
if (!systemApp) {
systemApp = await AppModel.create({
name: 'system',
scopes: ['crons', 'system', 'dashboard'],
client_id: createRandomString(12, 12),
client_secret: createRandomString(24, 24),
});
} else if (!systemApp.scopes.includes('dashboard')) {
await AppModel.update(
{ scopes: [...systemApp.scopes, 'dashboard'] },
{ where: { name: 'system' } },
);
}
await openService.initializeSystemApp();
const [systemConfig] = await SystemModel.findOrCreate({
where: { type: AuthDataType.systemConfig },
});
@@ -143,7 +126,11 @@ export default async () => {
// 清空所有运行中的实例记录(服务重启后进程已不存在)
await RunningInstanceModel.update(
{ status: InstanceStatus.stopped },
{
status: InstanceStatus.stopped,
finished_at: Math.floor(Date.now() / 1000),
// The previous process's exit code cannot be recovered after a restart.
},
{ where: { status: InstanceStatus.running } },
);
@@ -236,14 +223,15 @@ export default async () => {
} catch { }
// 初始化保存一次ck和定时任务数据
await cronService.autosave_crontab();
cronClient.readiness.configure(async () => {
await cronService.autosave_crontab(true);
await systemService.restoreCronConcurrency();
});
await cronClient.readiness.recover();
await envService.set_envs();
const authInfo = await userService.getAuthInfo();
const apps = await openService.findApps();
await shareStore.updateAuthInfo(authInfo);
if (apps?.length) {
await shareStore.updateApps(apps);
}
await openService.refreshApps();
};
+10 -1
View File
@@ -7,6 +7,8 @@ import config from '../config';
import { fileExist } from '../config/util';
import { join } from 'path';
import { t } from '../shared/i18n';
import Logger from './logger';
import { assertSubscriptionAlias } from '../shared/subscriptionPath';
export default async () => {
const systemService = Container.get(SystemService);
@@ -24,6 +26,7 @@ export default async () => {
if (await fileExist(tokenFile)) {
tokenCommand = `node ${tokenFile}`;
}
tokenCommand += ' --renew';
const cron = {
id: NaN,
name: t('生成token'),
@@ -70,6 +73,12 @@ export default async () => {
await subscriptionService.setSshConfig();
const subs = await subscriptionService.list();
for (const sub of subs) {
subscriptionService.handleTask(sub.get({ plain: true }), !sub.is_disabled);
try {
const doc = sub.get({ plain: true });
assertSubscriptionAlias(doc.alias);
await subscriptionService.handleTask(doc, !sub.is_disabled);
} catch (error) {
Logger.warn('跳过订阅 %s 的无效定时配置: %s', sub.id, error);
}
}
};
+87 -21
View File
@@ -1,4 +1,4 @@
import sockJs from 'sockjs';
import sockJs, { Connection } from 'sockjs';
import { Server } from 'http';
import { Container } from 'typedi';
import SockService from '../services/sock';
@@ -6,6 +6,7 @@ import { getPlatform } from '../config/util';
import { shareStore } from '../shared/store';
import { isValidToken } from '../shared/auth';
import config from '../config';
import jwt from 'jsonwebtoken';
export default async ({ server }: { server: Server }) => {
const echo = sockJs.createServer({
@@ -13,6 +14,65 @@ export default async ({ server }: { server: Server }) => {
log: () => {},
});
const sockService = Container.get(SockService);
const sessions = new Map<Connection, { token: string; platform: string }>();
let sessionTimer: ReturnType<typeof setInterval> | undefined;
let checking = false;
const validateSession = async (
authInfo: any,
token: string,
platform: string,
) => {
if (isValidToken(authInfo, token, platform, config.jwt.secret)) return true;
const claims = jwt.decode(token);
if (
!claims ||
typeof claims === 'string' ||
!Number.isSafeInteger(claims.userId) ||
claims.userId <= 0
)
return false;
const { default: UserManagementService } = await import(
'../services/userManagement'
);
const account = await Container.get(UserManagementService).validateToken(
token,
platform,
);
return account?.role === 0;
};
const checkSessions = async () => {
if (checking || sessions.size === 0) return;
checking = true;
// Connections accepted during this read must not use an older snapshot.
const batch = [...sessions];
try {
const current = await shareStore.getAuthInfo();
// Reuse validation only within this synchronous check of one auth snapshot.
const validated = new Map<string, Map<string, boolean>>();
for (const [conn, { token, platform }] of batch) {
if (!sessions.has(conn)) continue;
let platforms = validated.get(token);
if (!platforms) {
platforms = new Map();
validated.set(token, platforms);
}
let valid = platforms.get(platform);
if (valid === undefined) {
valid = await validateSession(current, token, platform);
platforms.set(platform, valid);
}
if (!valid) conn.close('401');
}
} catch {
for (const [conn] of batch) {
if (sessions.has(conn)) conn.close('401');
}
} finally {
checking = false;
}
};
echo.on('connection', async (conn) => {
if (!conn.headers || !conn.url || !conn.pathname) {
@@ -20,35 +80,41 @@ export default async ({ server }: { server: Server }) => {
return;
}
const authInfo = await shareStore.getAuthInfo();
let closed = false;
conn.on('close', () => {
closed = true;
sessions.delete(conn);
sockService.removeClient(conn);
if (sessions.size === 0 && sessionTimer) {
clearInterval(sessionTimer);
sessionTimer = undefined;
}
});
let authInfo;
try {
authInfo = await shareStore.getAuthInfo();
} catch {
if (!closed) conn.close('401');
return;
}
if (closed) return;
const platform = getPlatform(conn.headers['user-agent'] || '') || 'desktop';
const headerToken = conn.url.replace(`${conn.pathname}?token=`, '');
if (isValidToken(authInfo, headerToken, platform, config.jwt.secret)) {
if (await validateSession(authInfo, headerToken, platform)) {
if (closed) return;
sockService.addClient(conn);
const checkSession = setInterval(async () => {
try {
const current = await shareStore.getAuthInfo();
if (
!isValidToken(current, headerToken, platform, config.jwt.secret)
) {
conn.close('401');
}
} catch {
conn.close('401');
}
}, 1000);
checkSession.unref();
sessions.set(conn, { token: headerToken, platform });
if (!sessionTimer) {
sessionTimer = setInterval(checkSessions, 1000);
sessionTimer.unref();
}
conn.on('data', (message) => {
conn.write(message);
});
conn.on('close', function () {
clearInterval(checkSession);
sockService.removeClient(conn);
});
return;
}
+14
View File
@@ -0,0 +1,14 @@
import { ErrorRequestHandler } from 'express';
import { UniqueConstraintError } from 'sequelize';
import { t } from '../shared/i18n';
// Handle the database constraint itself so concurrent writes are covered too.
const uniqueConstraintError: ErrorRequestHandler = (err, req, res, next) => {
if (!(err instanceof UniqueConstraintError)) return next(err);
// Constraint details can contain environment values or application credentials.
res
.status(409)
.json({ code: 409, message: t('资源已存在,请检查重复的名称或值') });
};
export default uniqueConstraintError;
+68
View File
@@ -0,0 +1,68 @@
import { Request, Response, NextFunction } from 'express';
import { Op } from 'sequelize';
import { EnvModel } from '../data/env';
import { CrontabModel } from '../data/cron';
import { SubscriptionModel } from '../data/subscription';
// Ordinary users are read-only until task workers and shared files are isolated.
export async function userAccess(
req: Request,
res: Response,
next: NextFunction,
) {
if (!req.user || req.user.role === 0) return next();
const userId = req.user.userId;
if (!Number.isSafeInteger(userId) || userId! <= 0)
return res.status(403).json({ code: 403, message: '需要有效用户会话' });
const pathname = req.path.replace(/\/$/, '') || '/';
if (
(req.method === 'GET' &&
['/user', '/system', '/health'].includes(pathname)) ||
(req.method === 'POST' && pathname === '/user/logout')
)
return next();
if (req.method === 'GET' && pathname === '/crons/views')
return res.json({ code: 200, data: [] });
const match = pathname.match(
/^\/(envs|crons|subscriptions)(?:\/([1-9][0-9]*))?$/,
);
if (req.method !== 'GET' || !match)
return res
.status(403)
.json({
code: 403,
message: '普通用户只读自己的数据,执行和系统修改由管理员负责',
});
try {
const models: Record<string, any> = {
envs: EnvModel,
crons: CrontabModel,
subscriptions: SubscriptionModel,
};
const model = models[match[1]];
const where: any = { userId };
if (match[2]) {
const id = Number(match[2]);
if (!Number.isSafeInteger(id))
return res.status(400).json({ code: 400, message: '无效 ID' });
where.id = id;
const doc = await model.findOne({ where });
return doc
? res.json({ code: 200, data: doc.get({ plain: true }) })
: res.status(404).json({ code: 404, message: '数据不存在' });
}
const search =
typeof req.query.searchValue === 'string'
? req.query.searchValue.slice(0, 1024)
: '';
if (search) where.name = { [Op.like]: `%${search}%` };
const docs = await model.findAll({ where, order: [['createdAt', 'DESC']] });
const data = docs.map((doc: any) => doc.get({ plain: true }));
return res.json({
code: 200,
data: match[1] === 'crons' ? { data, total: data.length } : data,
});
} catch (error) {
return next(error);
}
}
+6 -1
View File
@@ -152,6 +152,7 @@ enum NotificationMode {
ntfy = 19;
wxPusherBot = 20;
wxPusherSpt = 21;
wpush = 22;
}
message NotificationInfo {
@@ -247,6 +248,10 @@ message NotificationInfo {
optional string wxPusherBotUids = 68;
optional string wxPusherSptList = 70;
optional string wpushApiKey = 71;
optional string wpushChannel = 72;
optional string wpushTopicCode = 73;
}
message SystemNotifyRequest {
@@ -275,4 +280,4 @@ service Api {
rpc EnableCrons(EnableCronsRequest) returns (Response) {}
rpc DisableCrons(DisableCronsRequest) returns (Response) {}
rpc RunCrons(RunCronsRequest) returns (Response) {}
}
}
+61 -1
View File
@@ -1,7 +1,7 @@
// Code generated by protoc-gen-ts_proto. DO NOT EDIT.
// versions:
// protoc-gen-ts_proto v2.6.1
// protoc v3.21.12
// protoc v3.17.3
// source: back/protos/api.proto
/* eslint-disable */
@@ -44,6 +44,7 @@ export enum NotificationMode {
ntfy = 19,
wxPusherBot = 20,
wxPusherSpt = 21,
wpush = 22,
UNRECOGNIZED = -1,
}
@@ -115,6 +116,9 @@ export function notificationModeFromJSON(object: any): NotificationMode {
case 21:
case "wxPusherSpt":
return NotificationMode.wxPusherSpt;
case 22:
case "wpush":
return NotificationMode.wpush;
case -1:
case "UNRECOGNIZED":
default:
@@ -168,6 +172,8 @@ export function notificationModeToJSON(object: NotificationMode): string {
return "wxPusherBot";
case NotificationMode.wxPusherSpt:
return "wxPusherSpt";
case NotificationMode.wpush:
return "wpush";
case NotificationMode.UNRECOGNIZED:
default:
return "UNRECOGNIZED";
@@ -400,6 +406,9 @@ export interface NotificationInfo {
wxPusherBotTopicIds?: string | undefined;
wxPusherBotUids?: string | undefined;
wxPusherSptList?: string | undefined;
wpushApiKey?: string | undefined;
wpushChannel?: string | undefined;
wpushTopicCode?: string | undefined;
}
export interface SystemNotifyRequest {
@@ -2967,6 +2976,9 @@ function createBaseNotificationInfo(): NotificationInfo {
wxPusherBotTopicIds: undefined,
wxPusherBotUids: undefined,
wxPusherSptList: undefined,
wpushApiKey: undefined,
wpushChannel: undefined,
wpushTopicCode: undefined,
};
}
@@ -3182,6 +3194,15 @@ export const NotificationInfo: MessageFns<NotificationInfo> = {
if (message.wxPusherSptList !== undefined) {
writer.uint32(562).string(message.wxPusherSptList);
}
if (message.wpushApiKey !== undefined) {
writer.uint32(570).string(message.wpushApiKey);
}
if (message.wpushChannel !== undefined) {
writer.uint32(578).string(message.wpushChannel);
}
if (message.wpushTopicCode !== undefined) {
writer.uint32(586).string(message.wpushTopicCode);
}
return writer;
},
@@ -3752,6 +3773,30 @@ export const NotificationInfo: MessageFns<NotificationInfo> = {
message.wxPusherSptList = reader.string();
continue;
}
case 71: {
if (tag !== 570) {
break;
}
message.wpushApiKey = reader.string();
continue;
}
case 72: {
if (tag !== 578) {
break;
}
message.wpushChannel = reader.string();
continue;
}
case 73: {
if (tag !== 586) {
break;
}
message.wpushTopicCode = reader.string();
continue;
}
}
if ((tag & 7) === 4 || tag === 0) {
break;
@@ -3845,6 +3890,9 @@ export const NotificationInfo: MessageFns<NotificationInfo> = {
: undefined,
wxPusherBotUids: isSet(object.wxPusherBotUids) ? globalThis.String(object.wxPusherBotUids) : undefined,
wxPusherSptList: isSet(object.wxPusherSptList) ? globalThis.String(object.wxPusherSptList) : undefined,
wpushApiKey: isSet(object.wpushApiKey) ? globalThis.String(object.wpushApiKey) : undefined,
wpushChannel: isSet(object.wpushChannel) ? globalThis.String(object.wpushChannel) : undefined,
wpushTopicCode: isSet(object.wpushTopicCode) ? globalThis.String(object.wpushTopicCode) : undefined,
};
},
@@ -4060,6 +4108,15 @@ export const NotificationInfo: MessageFns<NotificationInfo> = {
if (message.wxPusherSptList !== undefined) {
obj.wxPusherSptList = message.wxPusherSptList;
}
if (message.wpushApiKey !== undefined) {
obj.wpushApiKey = message.wpushApiKey;
}
if (message.wpushChannel !== undefined) {
obj.wpushChannel = message.wpushChannel;
}
if (message.wpushTopicCode !== undefined) {
obj.wpushTopicCode = message.wpushTopicCode;
}
return obj;
},
@@ -4138,6 +4195,9 @@ export const NotificationInfo: MessageFns<NotificationInfo> = {
message.wxPusherBotTopicIds = object.wxPusherBotTopicIds ?? undefined;
message.wxPusherBotUids = object.wxPusherBotUids ?? undefined;
message.wxPusherSptList = object.wxPusherSptList ?? undefined;
message.wpushApiKey = object.wpushApiKey ?? undefined;
message.wpushChannel = object.wpushChannel ?? undefined;
message.wpushTopicCode = object.wpushTopicCode ?? undefined;
return message;
},
};
+8 -1
View File
@@ -5,6 +5,7 @@ package com.ql.cron;
service Cron {
rpc addCron(AddCronRequest) returns (AddCronResponse);
rpc delCron(DeleteCronRequest) returns (DeleteCronResponse);
rpc setConcurrency(SetConcurrencyRequest) returns (AddCronResponse);
}
message ISchedule { string schedule = 1; }
@@ -17,10 +18,16 @@ message ICron {
string name = 5;
}
message AddCronRequest { repeated ICron crons = 1; }
message AddCronRequest {
repeated ICron crons = 1;
bool replace = 2;
}
message AddCronResponse {}
message DeleteCronRequest { repeated string ids = 1; }
message DeleteCronResponse {}
// Zero restores the default limit.
message SetConcurrencyRequest { int32 concurrency = 1; }
+110 -3
View File
@@ -1,7 +1,7 @@
// Code generated by protoc-gen-ts_proto. DO NOT EDIT.
// versions:
// protoc-gen-ts_proto v2.6.1
// protoc v3.21.12
// protoc v5.28.3
// source: back/protos/cron.proto
/* eslint-disable */
@@ -35,6 +35,7 @@ export interface ICron {
export interface AddCronRequest {
crons: ICron[];
replace: boolean;
}
export interface AddCronResponse {
@@ -47,6 +48,11 @@ export interface DeleteCronRequest {
export interface DeleteCronResponse {
}
/** Zero restores the default limit. */
export interface SetConcurrencyRequest {
concurrency: number;
}
function createBaseISchedule(): ISchedule {
return { schedule: "" };
}
@@ -232,7 +238,7 @@ export const ICron: MessageFns<ICron> = {
};
function createBaseAddCronRequest(): AddCronRequest {
return { crons: [] };
return { crons: [], replace: false };
}
export const AddCronRequest: MessageFns<AddCronRequest> = {
@@ -240,6 +246,9 @@ export const AddCronRequest: MessageFns<AddCronRequest> = {
for (const v of message.crons) {
ICron.encode(v!, writer.uint32(10).fork()).join();
}
if (message.replace !== false) {
writer.uint32(16).bool(message.replace);
}
return writer;
},
@@ -258,6 +267,14 @@ export const AddCronRequest: MessageFns<AddCronRequest> = {
message.crons.push(ICron.decode(reader, reader.uint32()));
continue;
}
case 2: {
if (tag !== 16) {
break;
}
message.replace = reader.bool();
continue;
}
}
if ((tag & 7) === 4 || tag === 0) {
break;
@@ -268,7 +285,10 @@ export const AddCronRequest: MessageFns<AddCronRequest> = {
},
fromJSON(object: any): AddCronRequest {
return { crons: globalThis.Array.isArray(object?.crons) ? object.crons.map((e: any) => ICron.fromJSON(e)) : [] };
return {
crons: globalThis.Array.isArray(object?.crons) ? object.crons.map((e: any) => ICron.fromJSON(e)) : [],
replace: isSet(object.replace) ? globalThis.Boolean(object.replace) : false,
};
},
toJSON(message: AddCronRequest): unknown {
@@ -276,6 +296,9 @@ export const AddCronRequest: MessageFns<AddCronRequest> = {
if (message.crons?.length) {
obj.crons = message.crons.map((e) => ICron.toJSON(e));
}
if (message.replace !== false) {
obj.replace = message.replace;
}
return obj;
},
@@ -285,6 +308,7 @@ export const AddCronRequest: MessageFns<AddCronRequest> = {
fromPartial<I extends Exact<DeepPartial<AddCronRequest>, I>>(object: I): AddCronRequest {
const message = createBaseAddCronRequest();
message.crons = object.crons?.map((e) => ICron.fromPartial(e)) || [];
message.replace = object.replace ?? false;
return message;
},
};
@@ -433,6 +457,64 @@ export const DeleteCronResponse: MessageFns<DeleteCronResponse> = {
},
};
function createBaseSetConcurrencyRequest(): SetConcurrencyRequest {
return { concurrency: 0 };
}
export const SetConcurrencyRequest: MessageFns<SetConcurrencyRequest> = {
encode(message: SetConcurrencyRequest, writer: BinaryWriter = new BinaryWriter()): BinaryWriter {
if (message.concurrency !== 0) {
writer.uint32(8).int32(message.concurrency);
}
return writer;
},
decode(input: BinaryReader | Uint8Array, length?: number): SetConcurrencyRequest {
const reader = input instanceof BinaryReader ? input : new BinaryReader(input);
let end = length === undefined ? reader.len : reader.pos + length;
const message = createBaseSetConcurrencyRequest();
while (reader.pos < end) {
const tag = reader.uint32();
switch (tag >>> 3) {
case 1: {
if (tag !== 8) {
break;
}
message.concurrency = reader.int32();
continue;
}
}
if ((tag & 7) === 4 || tag === 0) {
break;
}
reader.skip(tag & 7);
}
return message;
},
fromJSON(object: any): SetConcurrencyRequest {
return { concurrency: isSet(object.concurrency) ? globalThis.Number(object.concurrency) : 0 };
},
toJSON(message: SetConcurrencyRequest): unknown {
const obj: any = {};
if (message.concurrency !== 0) {
obj.concurrency = Math.round(message.concurrency);
}
return obj;
},
create<I extends Exact<DeepPartial<SetConcurrencyRequest>, I>>(base?: I): SetConcurrencyRequest {
return SetConcurrencyRequest.fromPartial(base ?? ({} as any));
},
fromPartial<I extends Exact<DeepPartial<SetConcurrencyRequest>, I>>(object: I): SetConcurrencyRequest {
const message = createBaseSetConcurrencyRequest();
message.concurrency = object.concurrency ?? 0;
return message;
},
};
export type CronService = typeof CronService;
export const CronService = {
addCron: {
@@ -453,11 +535,21 @@ export const CronService = {
responseSerialize: (value: DeleteCronResponse) => Buffer.from(DeleteCronResponse.encode(value).finish()),
responseDeserialize: (value: Buffer) => DeleteCronResponse.decode(value),
},
setConcurrency: {
path: "/com.ql.cron.Cron/setConcurrency",
requestStream: false,
responseStream: false,
requestSerialize: (value: SetConcurrencyRequest) => Buffer.from(SetConcurrencyRequest.encode(value).finish()),
requestDeserialize: (value: Buffer) => SetConcurrencyRequest.decode(value),
responseSerialize: (value: AddCronResponse) => Buffer.from(AddCronResponse.encode(value).finish()),
responseDeserialize: (value: Buffer) => AddCronResponse.decode(value),
},
} as const;
export interface CronServer extends UntypedServiceImplementation {
addCron: handleUnaryCall<AddCronRequest, AddCronResponse>;
delCron: handleUnaryCall<DeleteCronRequest, DeleteCronResponse>;
setConcurrency: handleUnaryCall<SetConcurrencyRequest, AddCronResponse>;
}
export interface CronClient extends Client {
@@ -491,6 +583,21 @@ export interface CronClient extends Client {
options: Partial<CallOptions>,
callback: (error: ServiceError | null, response: DeleteCronResponse) => void,
): ClientUnaryCall;
setConcurrency(
request: SetConcurrencyRequest,
callback: (error: ServiceError | null, response: AddCronResponse) => void,
): ClientUnaryCall;
setConcurrency(
request: SetConcurrencyRequest,
metadata: Metadata,
callback: (error: ServiceError | null, response: AddCronResponse) => void,
): ClientUnaryCall;
setConcurrency(
request: SetConcurrencyRequest,
metadata: Metadata,
options: Partial<CallOptions>,
callback: (error: ServiceError | null, response: AddCronResponse) => void,
): ClientUnaryCall;
}
export const CronClient = makeGenericClientConstructor(CronService, "com.ql.cron.Cron") as unknown as {
+1 -1
View File
@@ -1,7 +1,7 @@
// Code generated by protoc-gen-ts_proto. DO NOT EDIT.
// versions:
// protoc-gen-ts_proto v2.6.1
// protoc v3.21.12
// protoc v3.17.3
// source: back/protos/health.proto
/* eslint-disable */
+59 -84
View File
@@ -1,28 +1,15 @@
import { ServerUnaryCall, sendUnaryData, status } from '@grpc/grpc-js';
import { AddCronRequest, AddCronResponse } from '../protos/cron';
import nodeSchedule from 'node-schedule';
import { createCronJob, CronJob } from '../shared/cronScheduler';
import { isValidCronSchedule } from '../shared/cronSchedule';
import { scheduleStacks } from './data';
import { runCron } from '../shared/runCron';
import Logger from '../loaders/logger';
import { tf } from '../shared/i18n';
/**
* 预校验 cron 表达式,检测 node-schedule 会拒绝但 cron-parser 会接受的 pattern。
* node-schedule 对 bare /N(字段以 / 开头,如前无星号/数字前缀的 /6)返回 null,
* 提前拦截避免走 scheduleJob 后才发现无效。
*/
// Validate the entire batch before replacing any existing jobs.
const isValidCronField = (cron: string): boolean => {
// 检测 bare /N 模式:字段以 / 开头如 "/6",或空格后紧跟 "/6"
// node-schedule 会对这种字段返回 null
if (/\s\/\d/.test(cron) || /^\/\d/.test(cron)) {
return false;
}
// 检测 ? 字符:Quartz cron 语法,node-schedule 在大多数位置返回 null
// cron-parser 接受但 node-schedule 拒绝,提前拦截
if (/\?/.test(cron)) {
return false;
}
return true;
return isValidCronSchedule(cron);
};
const addCron = (
@@ -37,11 +24,7 @@ const addCron = (
if (!isValidCronField(schedule)) {
validationErrors.push(
tf(
'任务ID %s: 无效的 cron 表达式 "%s"(不支持裸 /N 步长和 ? 字符)',
String(id),
schedule,
),
tf('任务ID %s: 无效的 cron 表达式 "%s"', String(id), schedule),
);
}
@@ -50,7 +33,7 @@ const addCron = (
if (!isValidCronField(x.schedule)) {
validationErrors.push(
tf(
'任务ID %s (extra_schedule): 无效的 cron 表达式 "%s"(不支持裸 /N 步长和 ? 字符)',
'任务ID %s (extra_schedule): 无效的 cron 表达式 "%s"',
String(id),
x.schedule,
),
@@ -69,70 +52,62 @@ const addCron = (
return;
}
// ===== 第二遍:注册所有任务 =====
for (const item of call.request.crons) {
const { id, schedule, command, extra_schedules, name } = item;
// 取消该 id 已有的旧任务
if (scheduleStacks.has(id)) {
scheduleStacks.get(id)?.forEach((x) => x.cancel());
}
Logger.info(
'[schedule][创建定时任务] 任务ID: %s, 名称: %s, cron: %s, 执行命令: %s',
id,
name,
schedule,
command,
);
if (extra_schedules?.length) {
extra_schedules.forEach((x) => {
Logger.info(
'[schedule][创建定时任务] 任务ID: %s, 名称: %s, cron: %s, 执行命令: %s',
id,
name,
x.schedule,
command,
// Prepare stopped jobs first: construction errors must preserve the old snapshot.
const prepared = new Map<string, CronJob[]>();
try {
for (const item of call.request.crons) {
const jobs: CronJob[] = [];
prepared.get(item.id)?.forEach((job) => job.cancel());
prepared.set(item.id, jobs);
for (const schedule of [
item.schedule,
...(item.extra_schedules || []).map((x) => x.schedule),
]) {
jobs.push(
createCronJob(
schedule,
async () => {
Logger.info('[schedule][准备运行任务] 命令: %s', item.command);
await runCron(
item.command,
item,
() => scheduleStacks.get(item.id) === jobs,
);
},
{
name: `${item.id}: ${item.name || ''}`,
logger: Logger,
start: false,
},
),
);
});
}
}
} catch (error) {
for (const jobs of prepared.values()) jobs.forEach((job) => job.cancel());
const err: any = new Error(
error instanceof Error ? error.message : String(error),
);
err.code = status.INVALID_ARGUMENT;
err.details = err.message;
callback(err, null);
return;
}
const mainJob = nodeSchedule.scheduleJob(id, schedule, async () => {
Logger.info(`[schedule][准备运行任务] 命令: ${command}`);
runCron(command, item);
});
if (!mainJob) {
Logger.warn(
'[schedule][创建定时任务] scheduleJob 返回 null(不符合预期,已通过预校验): 任务ID: %s, cron: %s',
id,
schedule,
);
}
const extraJobs = extra_schedules?.length
? extra_schedules.map((x) => {
const job = nodeSchedule.scheduleJob(id, x.schedule, async () => {
Logger.info(`[schedule][准备运行任务] 命令: ${command}`);
runCron(command, item);
});
if (!job) {
Logger.warn(
'[schedule][创建定时任务] scheduleJob 返回 null(不符合预期,已通过预校验): 任务ID: %s, cron: %s',
id,
x.schedule,
);
}
return job;
})
: [];
// 过滤 null(兜底保护,正常情况下预校验已拦截)
const jobs = [mainJob, ...extraJobs].filter((x) => x != null);
if (jobs.length > 0) {
scheduleStacks.set(id, jobs);
}
if (call.request.replace) {
for (const jobs of scheduleStacks.values())
jobs.forEach((job) => job?.cancel());
scheduleStacks.clear();
}
for (const [id, jobs] of prepared) {
scheduleStacks.get(id)?.forEach((job) => job.cancel());
scheduleStacks.set(id, jobs);
jobs.forEach((job) => job.start());
Logger.info(
'[schedule][创建定时任务] 任务ID: %s, 规则数: %s',
id,
jobs.length,
);
}
callback(null, null);
+10 -6
View File
@@ -38,6 +38,7 @@ import {
RunCronsRequest,
} from '../protos/api';
import { NotificationInfo } from '../data/notify';
import { Model } from 'sequelize';
Container.set('logger', LoggerInstance);
@@ -247,13 +248,16 @@ export const systemNotify = async (
const normalizeCronData = (data: CronItem | null): CronItem | undefined => {
if (!data) return undefined;
// create() returns a Sequelize instance; spreading it omits attribute getters.
const cron = data instanceof Model ? (data.get({ plain: true }) as CronItem) : data;
return {
...data,
sub_id: data.sub_id ?? undefined,
extra_schedules: data.extra_schedules ?? [],
pid: data.pid ?? undefined,
task_before: data.task_before ?? undefined,
task_after: data.task_after ?? undefined,
...cron,
labels: cron.labels ?? [],
sub_id: cron.sub_id ?? undefined,
extra_schedules: cron.extra_schedules ?? [],
pid: cron.pid ?? undefined,
task_before: cron.task_before ?? undefined,
task_after: cron.task_after ?? undefined,
};
};
+90 -11
View File
@@ -1,4 +1,4 @@
import { credentials } from '@grpc/grpc-js';
import { credentials, status, Metadata } from '@grpc/grpc-js';
import {
AddCronRequest,
AddCronResponse,
@@ -9,7 +9,41 @@ import {
import config from '../config';
import { getGrpcCerts } from '../config/grpcCerts';
import { HealthService } from '../protos/health';
import { SchedulerReadiness } from '../shared/schedulerReadiness';
class Client {
readonly readiness = new SchedulerReadiness(() => this.probe());
private async waitForReady(timeoutMs: number) {
try {
await new Promise<void>((resolve, reject) => {
this.client.waitForReady(Date.now() + timeoutMs, (err) =>
err ? reject(err) : resolve()
);
});
} catch (error) {
this.readiness.invalidate();
throw Object.assign(
error instanceof Error ? error : new Error(String(error)),
{ status: 503 }
);
}
}
private async probe(): Promise<void> {
await this.waitForReady(1000);
await new Promise<void>((resolve, reject) => {
this.client.makeUnaryRequest(
HealthService.check.path,
HealthService.check.requestSerialize,
HealthService.check.responseDeserialize,
{ service: 'scheduler' },
{ deadline: Date.now() + 1000 },
(err, res) => err ? reject(err) : res?.status === 1 ? resolve() : reject(new Error('Scheduler unavailable')),
);
});
}
private _client: CronClient | null = null;
private get client(): CronClient {
@@ -28,22 +62,67 @@ class Client {
return this._client;
}
addCron(request: AddCronRequest['crons']): Promise<AddCronResponse> {
async setConcurrency(concurrency: number): Promise<void> {
await this.waitForReady(2000);
return new Promise((resolve, reject) => {
this.client.addCron({ crons: request }, (err, res) => {
if (err) {
reject(err);
}
resolve(res);
});
this.client.setConcurrency(
{ concurrency }, new Metadata(), { deadline: Date.now() + 5000 },
(err) => {
if (err) {
if (err.code === status.UNAVAILABLE || err.code === status.DEADLINE_EXCEEDED) {
this.readiness.invalidate();
Object.assign(err, { status: 503 });
}
return reject(err);
}
resolve();
},
);
});
}
delCron(request: DeleteCronRequest['ids']): Promise<DeleteCronResponse> {
async addCron(
request: AddCronRequest['crons'],
replace = false
): Promise<AddCronResponse> {
await this.waitForReady(2000);
// Include every rule in a recovery snapshot, but bound a stuck worker.
const ruleCount = request.reduce(
(count, item) => count + 1 + (item.extra_schedules?.length || 0),
0,
);
const registrationTimeoutMs = Math.min(120000, 5000 + ruleCount * 5);
return new Promise((resolve, reject) => {
this.client.delCron({ ids: request }, (err, res) => {
this.client.addCron(
{ crons: request, replace },
new Metadata(),
{ deadline: Date.now() + registrationTimeoutMs },
(err, res) => {
if (err) {
if (err.code === status.UNAVAILABLE || err.code === status.DEADLINE_EXCEEDED) {
// A timed-out write may already have reached the scheduler.
// Reconcile its state from the DB instead of replaying the RPC.
this.readiness.invalidate();
Object.assign(err, { status: 503 });
}
return reject(err);
}
resolve(res);
}
);
});
}
async delCron(request: DeleteCronRequest['ids']): Promise<DeleteCronResponse> {
await this.waitForReady(2000);
return new Promise((resolve, reject) => {
this.client.delCron({ ids: request }, new Metadata(), { deadline: Date.now() + 5000 }, (err, res) => {
if (err) {
reject(err);
if (err.code === status.UNAVAILABLE || err.code === status.DEADLINE_EXCEEDED) {
this.readiness.invalidate();
Object.assign(err, { status: 503 });
}
return reject(err);
}
resolve(res);
});
+2 -2
View File
@@ -1,6 +1,6 @@
import nodeSchedule from 'node-schedule';
import { CronJob } from '../shared/cronScheduler';
import { ToadScheduler } from 'toad-scheduler';
export const scheduleStacks = new Map<string, nodeSchedule.Job[]>();
export const scheduleStacks = new Map<string, CronJob[]>();
export const intervalSchedule = new ToadScheduler();
+1 -1
View File
@@ -13,7 +13,7 @@ const delCron = (
'[schedule][取消定时任务] 任务ID: %s',
id,
);
// 过滤掉 nodeSchedule.scheduleJob() 对无效表达式返回的 null,
// 防御性过滤历史调度栈中的空任务,
// 否则对 null 调 cancel() 会让整个取消流程抛出 UNKNOWN 错误,
// 进而导致 HTTP 端的 remove() 跳过 setCrontab(),造成 crontab.list 残留。
scheduleStacks.get(id)?.filter((x) => x != null).forEach((x) => {
+3
View File
@@ -52,6 +52,9 @@ const check = async (
callback: sendUnaryData<HealthCheckResponse>,
) => {
switch (call.request.service) {
// Local scheduler liveness only: never call HTTP from this probe.
case 'scheduler':
return callback(null, { status: 1 });
case 'cron': {
const healthUrl = `http://localhost:${config.port}${
config.baseUrl || ''
+26
View File
@@ -0,0 +1,26 @@
import { ServerUnaryCall, sendUnaryData, status } from '@grpc/grpc-js';
import os from 'os';
import { AddCronResponse, SetConcurrencyRequest } from '../protos/cron';
import taskLimit from '../shared/pLimit';
export async function setConcurrency(
call: ServerUnaryCall<SetConcurrencyRequest, AddCronResponse>,
callback: sendUnaryData<AddCronResponse>,
) {
const limit = call.request.concurrency;
if (!Number.isInteger(limit) || limit < 0 || limit > 2147483647) {
callback(
Object.assign(new Error('Invalid concurrency'), {
code: status.INVALID_ARGUMENT,
}),
null,
);
return;
}
try {
await taskLimit.setCustomLimit(limit || Math.max(os.cpus().length, 4));
callback(null, {});
} catch (error) {
callback(error as Error, null);
}
}
+3 -4
View File
@@ -12,13 +12,12 @@ export default class ConfigService {
public async getFile(filePath: string, res: Response) {
let content = '';
if (!filePath) {
if (!filePath || filePath.startsWith('data/scripts/')) {
return res.send({ code: 403, message: t('文件无法访问') });
}
const scriptFile = filePath.startsWith('data/scripts/');
const resolved = resolveFileAccess(
scriptFile ? config.scriptPath : config.configPath,
[scriptFile ? filePath.slice('data/scripts/'.length) : filePath],
config.configPath,
[filePath],
config.blackFileList,
);
if (!resolved) {
+412 -269
View File
@@ -1,6 +1,13 @@
import { randomUUID } from 'crypto';
import { getInvalidCronSchedules, isValidCronSchedule } from '../shared/cronSchedule';
import {
withSchedulerMutation,
schedulerRegistrationError,
} from '../shared/schedulerMutationLock';
import { Service, Inject } from 'typedi';
import winston from 'winston';
import config from '../config';
import { resolveLogPath } from '../shared/logPath';
import { Crontab, CrontabModel, CrontabStatus } from '../data/cron';
import {
RunningInstanceModel,
@@ -8,12 +15,10 @@ import {
} from '../data/runningInstance';
import { exec, execSync } from 'child_process';
import fs from 'fs/promises';
import CronExpressionParser from 'cron-parser';
import {
getFileContentByName,
fileExist,
killTask,
killAllTasks,
getUniqPath,
safeJSONParse,
isDemoEnv,
@@ -31,8 +36,10 @@ import { writeFileWithLock } from '../shared/utils';
import { t } from '../shared/i18n';
import { ScheduleType } from '../interface/schedule';
import { logStreamManager } from '../shared/logStreamManager';
import { observeChildProcess, asError } from '../shared/childProcess';
import { isEmpty } from 'lodash';
import { LogReadOptions, readLogChunk } from '../shared/logReader';
import { resolveFileAccess } from '../shared/fileAccess';
@Service()
export default class CronService {
@@ -40,10 +47,21 @@ export default class CronService {
private isNodeCron(cron: Crontab) {
const { schedule, extra_schedules } = cron;
if (Number(schedule?.split(/ +/).length) > 5 || extra_schedules?.length) {
return true;
}
return false;
const fields = schedule?.trim().split(/\s+/) || [];
// System crontab only receives portable numeric five-field expressions.
// Extended syntax, macros and legacy shorthand use the Node scheduler.
return (
fields.length !== 5 ||
/[^\d\s*,/\-]/.test(schedule || '') ||
Boolean(extra_schedules?.length) ||
// BusyBox treats N/step as a single value and doesn't support Sunday=7.
fields.some((field) => field.split(',').some((part) => /^\d+\//.test(part))) ||
fields[4].includes('7') ||
// BusyBox steps DOM from zero; its full-range day/week wildcard
// handling also differs from the legacy Node parser's OR semantics.
fields[2].includes('/') ||
(fields[2] !== '*' && fields[4] !== '*')
);
}
private get schedulerMode(): 'system' | 'node' {
@@ -97,42 +115,45 @@ export default class CronService {
}
public async create(payload: Crontab): Promise<Crontab> {
const tab = new Crontab(payload);
tab.saved = false;
tab.log_name = await this.getLogName(tab);
const doc = await this.insert(tab);
return withSchedulerMutation(async () => {
const tab = new Crontab(payload);
tab.saved = false;
tab.log_name = await this.getLogName(tab);
const doc = await this.insert(tab);
if (isDemoEnv()) {
return doc;
}
if (this.shouldUseCronClient(doc)) {
try {
await cronClient.addCron([
{
name: doc.name || '',
id: String(doc.id),
schedule: doc.schedule!,
command: this.makeCommand(doc),
extra_schedules: doc.extra_schedules || [],
},
]);
} catch (error: any) {
// gRPC 注册失败时回滚 DB 记录,避免产生"僵尸任务"
// (DB 和 crontab.list 有记录但调度器永远不会执行)
await CrontabModel.destroy({ where: { id: doc.id } });
this.logger.error(
'[crontab] Failed to register cron job in scheduler, task creation rolled back:',
error?.message || error,
);
throw new Error(
`${t('调度器注册失败,任务创建已回滚')}: ${(error as any)?.details || error?.message}`,
);
if (isDemoEnv()) {
return doc;
}
}
await this.setCrontab();
return doc;
if (this.shouldUseCronClient(doc)) {
try {
await cronClient.addCron([
{
name: doc.name || '',
id: String(doc.id),
schedule: doc.schedule!,
command: this.makeCommand(doc),
extra_schedules: doc.extra_schedules || [],
},
]);
} catch (error: any) {
// gRPC 注册失败时回滚 DB 记录,避免产生"僵尸任务"
// (DB 和 crontab.list 有记录但调度器永远不会执行)
await CrontabModel.destroy({ where: { id: doc.id } });
this.logger.error(
'[crontab] Failed to register cron job in scheduler, task creation rolled back:',
error?.message || error,
);
throw schedulerRegistrationError(
`${t('调度器注册失败,任务创建已回滚')}: ${(error as any)?.details || error?.message}`,
error,
);
}
}
await this.setCrontab();
return doc;
});
}
public async insert(payload: Crontab): Promise<Crontab> {
@@ -140,69 +161,68 @@ export default class CronService {
}
public async update(payload: Partial<Crontab>): Promise<Crontab> {
const doc = await this.getDb({ id: payload.id });
const tab = new Crontab({ ...doc, ...payload });
tab.saved = false;
tab.log_name = await this.getLogName(tab);
const newDoc = await this.updateDb(tab);
if (doc.isDisabled === 1 || isDemoEnv()) {
return newDoc;
}
try {
await cronClient.delCron([String(newDoc.id)]);
} catch (error: any) {
this.logger.warn(
'[crontab] Failed to unregister cron job in scheduler:',
error?.message || error,
);
}
if (this.shouldUseCronClient(newDoc)) {
try {
await cronClient.addCron([
{
name: doc.name || '',
id: String(newDoc.id),
schedule: newDoc.schedule!,
command: this.makeCommand(newDoc),
extra_schedules: newDoc.extra_schedules || [],
},
]);
} catch (error: any) {
// gRPC 注册新任务失败 → 回滚 DB 到旧数据,并尝试恢复旧调度注册
await CrontabModel.update(doc, { where: { id: doc.id } });
if (this.shouldUseCronClient(doc)) {
try {
await cronClient.addCron([
{
name: doc.name || '',
id: String(doc.id),
schedule: doc.schedule!,
command: this.makeCommand(doc),
extra_schedules: doc.extra_schedules || [],
},
]);
} catch (_recoveryError: any) {
this.logger.warn(
'[crontab] Failed to restore old cron job in scheduler after rollback:',
_recoveryError?.message || _recoveryError,
);
}
}
this.logger.error(
'[crontab] Failed to register updated cron job in scheduler, update rolled back:',
error?.message || error,
);
throw new Error(
`${t('调度器注册失败,任务更新已回滚')}: ${(error as any)?.details || error?.message}`,
);
return withSchedulerMutation(async () => {
const doc = await this.getDb({ id: payload.id });
const tab = new Crontab({ ...doc, ...payload });
tab.saved = false;
tab.log_name = await this.getLogName(tab);
if (doc.isDisabled === 1 || isDemoEnv()) {
return await this.updateDb(tab);
}
}
await this.setCrontab();
return newDoc;
// Keep the DB snapshot unchanged if deletion has an uncertain outcome.
// Recovery uses that snapshot after this mutation releases its lock.
await cronClient.delCron([String(doc.id)]);
const newDoc = await this.updateDb(tab);
if (this.shouldUseCronClient(newDoc)) {
try {
await cronClient.addCron([
{
name: doc.name || '',
id: String(newDoc.id),
schedule: newDoc.schedule!,
command: this.makeCommand(newDoc),
extra_schedules: newDoc.extra_schedules || [],
},
]);
} catch (error: any) {
// gRPC 注册新任务失败 → 回滚 DB 到旧数据,并尝试恢复旧调度注册
await CrontabModel.update(omit(doc, ['queued_token']), {
where: { id: doc.id },
});
if (this.shouldUseCronClient(doc)) {
try {
await cronClient.addCron([
{
name: doc.name || '',
id: String(doc.id),
schedule: doc.schedule!,
command: this.makeCommand(doc),
extra_schedules: doc.extra_schedules || [],
},
]);
} catch (_recoveryError: any) {
this.logger.warn(
'[crontab] Failed to restore old cron job in scheduler after rollback:',
_recoveryError?.message || _recoveryError,
);
}
}
this.logger.error(
'[crontab] Failed to register updated cron job in scheduler, update rolled back:',
error?.message || error,
);
throw schedulerRegistrationError(
`${t('调度器注册失败,任务更新已回滚')}: ${(error as any)?.details || error?.message}`,
error,
);
}
}
await this.setCrontab();
return newDoc;
});
}
public async updateDb(payload: Crontab): Promise<Crontab> {
@@ -227,6 +247,9 @@ export default class CronService {
last_execution_time: number;
exit_code?: number;
}) {
if (log_path && log_path !== '/dev/null') {
resolveLogPath(config.logPath, log_path);
}
let options: any = {
status,
pid,
@@ -290,16 +313,11 @@ export default class CronService {
}
public async remove(ids: number[]) {
await CrontabModel.destroy({ where: { id: ids } });
try {
return withSchedulerMutation(async () => {
await cronClient.delCron(ids.map(String));
} catch (error: any) {
this.logger.warn(
'[crontab] Failed to unregister cron job in scheduler:',
error?.message || error,
);
}
await this.setCrontab();
await CrontabModel.destroy({ where: { id: ids } });
await this.setCrontab();
});
}
public async pin(ids: number[]) {
@@ -570,46 +588,99 @@ export default class CronService {
}
public async run(ids: number[]) {
const queuedToken = randomUUID();
await CrontabModel.update(
{ status: CrontabStatus.queued },
{ status: CrontabStatus.queued, queued_token: queuedToken },
{ where: { id: ids } },
);
ids.forEach((id) => {
this.runSingle(id);
this.runSingle(id, queuedToken);
});
}
public async stop(ids: number[]) {
const docs = await CrontabModel.findAll({ where: { id: ids } });
// Cancel the queued snapshot first, so a late spawn cannot claim it.
for (const doc of docs) {
// Kill all running instances of this task
try {
if (doc.pid) {
await killTask(doc.pid);
}
const command = doc.command.replace(/\s+/g, ' ').trim();
await killAllTasks(command);
this.logger.info(
`[panel][停止所有运行中的任务实例] 任务ID: ${doc.id}, 命令: ${command}`,
if (doc.status === CrontabStatus.queued) {
const [cancelled] = await CrontabModel.update(
{ status: CrontabStatus.idle, pid: null, queued_token: null } as any,
{
where: {
id: doc.id,
status: CrontabStatus.queued,
[Op.and]: [
where(colFn('log_path'), { [Op.eq]: doc.log_path ?? null }),
where(colFn('queued_token'), {
[Op.eq]: doc.queued_token ?? null,
}),
],
},
}
);
// A concurrent claim may have won; capture its PID before signalling.
if (!cancelled) await doc.reload();
}
}
const stoppingInstances = await RunningInstanceModel.findAll({
attributes: ['id', 'pid', 'cron_id'],
where: { cron_id: ids, status: InstanceStatus.running },
});
const targets = new Set<number>(
[
...stoppingInstances.map((instance) => instance.pid),
...docs
.filter((doc) => doc.status === CrontabStatus.running)
.map((doc) => doc.pid),
].filter((pid): pid is number => typeof pid === 'number' && pid > 0)
);
const stopped = new Set<number>();
for (const pid of targets) {
try {
await killTask(pid, true);
stopped.add(pid);
} catch (error) {
this.logger.error(
`[panel][停止任务失败] 任务ID: ${doc.id}, 错误: ${error}`,
'[panel][停止任务失败] PID: %s, 错误: %s',
pid,
asError(error).message
);
}
}
// Mark all running instances as stopped
const finishedAt = dayjs().unix();
await RunningInstanceModel.update(
{ status: InstanceStatus.stopped, finished_at: finishedAt },
{ where: { cron_id: ids, status: InstanceStatus.running } },
);
await CrontabModel.update(
{ status: CrontabStatus.idle, pid: undefined },
{ where: { id: ids } },
);
const stoppedIds = stoppingInstances
.filter((instance) => instance.pid && stopped.has(instance.pid))
.map((instance) => instance.id!);
if (stoppedIds.length) {
await RunningInstanceModel.update(
{ status: InstanceStatus.stopped, finished_at: dayjs().unix() },
{ where: { id: stoppedIds } }
);
}
for (const doc of docs) {
if (
doc.status !== CrontabStatus.running ||
(doc.pid && !stopped.has(doc.pid))
)
continue;
const remaining = await RunningInstanceModel.count({
where: { cron_id: doc.id, status: InstanceStatus.running },
});
if (remaining) continue;
await CrontabModel.update(
{ status: CrontabStatus.idle, pid: null, queued_token: null } as any,
{
where: {
id: doc.id,
status: CrontabStatus.running,
[Op.and]: [
where(colFn('queued_token'), { [Op.eq]: doc.queued_token ?? null }),
where(colFn('pid'), { [Op.eq]: doc.pid ?? null }),
where(colFn('log_path'), { [Op.eq]: doc.log_path ?? null }),
],
},
}
);
}
}
public async stopInstance(instanceId: number) {
@@ -646,125 +717,184 @@ export default class CronService {
return { code: 200, message: t('实例已停止') };
}
private async runSingle(cronId: number): Promise<number | void> {
return taskLimit.manualRunWithCronLimit(() => {
return new Promise(async (resolve: any) => {
private async runSingle(
cronId: number,
expectedToken?: string,
): Promise<number | void> {
return taskLimit.manualRunWithCronLimit(async () => {
let absolutePath: string | undefined;
let logPath: string | undefined;
let queuedLogPath: string | null | undefined;
let queuedToken: string | null = null;
let claimed = false;
try {
const cron = await this.getDb({ id: cronId });
const params = {
name: cron.name,
command: cron.command,
schedule: cron.schedule,
extra_schedules: cron.extra_schedules,
};
if (cron.status !== CrontabStatus.queued) {
resolve(params);
return;
}
this.logger.info(
`[panel][开始执行任务] 参数: ${JSON.stringify(params)}`,
);
let { id, command, log_name } = cron;
if (
cron.status !== CrontabStatus.queued ||
(expectedToken !== undefined && cron.queued_token !== expectedToken)
) return;
queuedToken = cron.queued_token ?? null;
queuedLogPath = cron.log_path ?? null;
const { id, command, log_name } = cron;
const uniqPath =
log_name === '/dev/null' || !log_name
? await getUniqPath(command, `${id}`)
: log_name;
const logTime = dayjs().format('YYYY-MM-DD-HH-mm-ss-SSS');
const logDirPath = path.resolve(config.logPath, `${uniqPath}`);
await fs.mkdir(logDirPath, { recursive: true });
const logPath = `${uniqPath}/${logTime}.log`;
const absolutePath = path.resolve(config.logPath, `${logPath}`);
logPath = `${uniqPath}/${logTime}.log`;
absolutePath = resolveFileAccess(config.logPath, [logPath]);
if (!absolutePath)
throw new Error('Log path is outside the log directory');
await fs.mkdir(path.dirname(absolutePath), { recursive: true });
const outputPath = absolutePath;
const cp = spawn(
`real_log_path=${logPath} no_delay=true ${this.makeCommand(
cron,
true,
true
)}`,
{ shell: '/bin/bash' },
{ shell: '/bin/bash' }
);
await CrontabModel.update(
{ status: CrontabStatus.running, pid: cp.pid, log_path: logPath },
{ where: { id } },
);
cp.stdout.on('data', async (data) => {
await logStreamManager.write(absolutePath, data.toString());
// Install observers before the first await: very short children may already exit.
const { completed } = observeChildProcess(cp, {
onStart: async () => {
try {
const [count] = await CrontabModel.update(
{
status: CrontabStatus.running,
pid: cp.pid,
log_path: logPath,
},
{
where: {
id,
status: CrontabStatus.queued,
[Op.and]: [
where(colFn('queued_token'), { [Op.eq]: queuedToken }),
where(colFn('log_path'), { [Op.eq]: queuedLogPath }),
],
},
}
);
if (count !== 1)
throw new Error(
'Task was stopped or superseded before startup'
);
claimed = true;
} catch (error) {
if (cp.pid) await killTask(cp.pid, true);
throw error;
}
},
onStdout: (message) => logStreamManager.write(outputPath, message),
onStderr: (message) => logStreamManager.write(outputPath, message),
});
cp.stderr.on('data', async (data) => {
this.logger.info(
'[panel][执行任务失败] 命令: %s, 错误信息: %j',
command,
data.toString(),
);
await logStreamManager.write(absolutePath, data.toString());
});
cp.on('error', async (err) => {
const result = await completed;
if (result.error) {
this.logger.error(
'[panel][创建任务失败] 命令: %s, 错误信息: %j',
command,
err,
'[panel][执行任务失败] 任务ID: %s, 错误: %s',
id,
result.error.message
);
await logStreamManager.write(absolutePath, JSON.stringify(err));
});
cp.on('exit', async (code) => {
this.logger.info(
'[panel][执行任务结束] 参数: %s, 退出码: %j',
JSON.stringify(params),
code,
}
this.logger.info(
'[panel][执行任务结束] 任务ID: %s, 退出码: %j',
id,
result.code
);
return { ...cron, pid: cp.pid, ...result } as any;
} catch (error) {
this.logger.error(
'[panel][创建任务失败] 任务ID: %s, 错误: %s',
cronId,
asError(error).message
);
} finally {
try {
if (absolutePath) await logStreamManager.closeStream(absolutePath);
} catch (error) {
this.logger.error(
'[panel][关闭任务日志失败] %s',
asError(error).message
);
await logStreamManager.closeStream(absolutePath);
resolve({ ...params, pid: cp.pid, code });
});
});
}
try {
// Do not overwrite a newer run's state or its script-reported exit code.
await CrontabModel.update(
{ status: CrontabStatus.idle, pid: null, queued_token: null } as any,
{
where: {
id: cronId,
[Op.and]: where(colFn('queued_token'), { [Op.eq]: queuedToken }),
[Op.or]: [
...(queuedLogPath !== undefined && !claimed
? [
{
status: CrontabStatus.queued,
[Op.and]: where(colFn('log_path'), {
[Op.eq]: queuedLogPath,
}),
},
]
: []),
...(claimed && logPath
? [{ log_path: logPath, status: CrontabStatus.running }]
: []),
],
},
}
);
} catch (error) {
this.logger.error(
'[panel][清理任务状态失败] %s',
asError(error).message
);
}
}
});
}
public async disabled(ids: number[]) {
await CrontabModel.update({ isDisabled: 1 }, { where: { id: ids } });
try {
return withSchedulerMutation(async () => {
await cronClient.delCron(ids.map(String));
} catch (error: any) {
this.logger.warn(
'[crontab] Failed to unregister cron job in scheduler:',
error?.message || error,
);
}
await this.setCrontab();
await CrontabModel.update({ isDisabled: 1 }, { where: { id: ids } });
await this.setCrontab();
});
}
public async enabled(ids: number[]) {
await CrontabModel.update({ isDisabled: 0 }, { where: { id: ids } });
const docs = await CrontabModel.findAll({ where: { id: ids } });
const crons = docs
.filter((x) => this.shouldUseCronClient(x))
.map((doc) => ({
name: doc.name || '',
id: String(doc.id),
schedule: doc.schedule!,
command: this.makeCommand(doc),
extra_schedules: doc.extra_schedules || [],
}));
return withSchedulerMutation(async () => {
await CrontabModel.update({ isDisabled: 0 }, { where: { id: ids } });
const docs = await CrontabModel.findAll({ where: { id: ids } });
const crons = docs
.filter((x) => this.shouldUseCronClient(x))
.map((doc) => ({
name: doc.name || '',
id: String(doc.id),
schedule: doc.schedule!,
command: this.makeCommand(doc),
extra_schedules: doc.extra_schedules || [],
}));
if (isDemoEnv()) {
return;
}
if (isDemoEnv()) {
return;
}
try {
await cronClient.addCron(crons);
} catch (error: any) {
// gRPC 注册失败 → 回滚启用状态,避免 DB 显示已启用但调度器未注册
await CrontabModel.update({ isDisabled: 1 }, { where: { id: ids } });
this.logger.error(
'[crontab] Failed to register cron job in scheduler, enable rolled back:',
error?.message || error,
);
throw new Error(
`${t('调度器注册失败,任务启用已回滚')}: ${(error as any)?.details || error?.message}`,
);
}
await this.setCrontab();
try {
await cronClient.addCron(crons);
} catch (error: any) {
// gRPC 注册失败 → 回滚启用状态,避免 DB 显示已启用但调度器未注册
await CrontabModel.update({ isDisabled: 1 }, { where: { id: ids } });
this.logger.error(
'[crontab] Failed to register cron job in scheduler, enable rolled back:',
error?.message || error,
);
throw schedulerRegistrationError(
`${t('调度器注册失败,任务启用已回滚')}: ${(error as any)?.details || error?.message}`,
error,
);
}
await this.setCrontab();
});
}
public async log(
@@ -799,7 +929,9 @@ export default class CronService {
truncated: false,
};
}
const absolutePath = path.resolve(config.logPath, `${doc.log_path}`);
const absolutePath = doc.log_path
? resolveLogPath(config.logPath, doc.log_path)
: '';
const logFileExist = doc.log_path && (await fileExist(absolutePath));
if (logFileExist) {
const chunk = await readLogChunk(`${absolutePath}`, options);
@@ -829,12 +961,12 @@ export default class CronService {
public async logs(id: number) {
const doc = await this.getDb({ id });
if (!doc || !doc.log_path) {
if (!doc || !doc.log_path || doc.log_path === '/dev/null') {
return [];
}
const relativeDir = path.dirname(`${doc.log_path}`);
const dir = path.resolve(config.logPath, relativeDir);
const dir = path.dirname(resolveLogPath(config.logPath, doc.log_path));
const dirExist = await fileExist(dir);
if (dirExist) {
let files = await fs.readdir(dir);
@@ -938,7 +1070,7 @@ export default class CronService {
if (
command &&
schedule &&
CronExpressionParser.parse(schedule).hasNext()
isValidCronSchedule(schedule)
) {
const name = namePrefix + '_' + index;
@@ -957,39 +1089,56 @@ export default class CronService {
});
}
public async autosave_crontab() {
const tabs = await this.crontabs();
const regularCrons = tabs.data
.filter(
(x) =>
x.isDisabled !== 1 &&
this.shouldUseCronClient(x),
)
.map((doc) => ({
name: doc.name || '',
id: String(doc.id),
schedule: doc.schedule!,
command: this.makeCommand(doc),
extra_schedules: doc.extra_schedules || [],
}));
public async autosave_crontab(requireScheduler = false) {
return withSchedulerMutation(async () => {
const tabs = await this.crontabs();
// A bad persisted rule must not block startup or all other schedules.
// Keep the DB row editable; omit it from both runtime scheduler snapshots.
tabs.data = tabs.data.filter((doc) => {
if (doc.isDisabled === 1) return true;
const invalidSchedules = getInvalidCronSchedules(doc);
if (invalidSchedules.length === 0) return true;
this.logger.warn(
'[crontab] Skipping task with invalid schedule (id=%s, name=%s): %s',
doc.id,
doc.name || '',
JSON.stringify(invalidSchedules),
);
return false;
});
const regularCrons = tabs.data
.filter(
(x) =>
x.isDisabled !== 1 &&
this.shouldUseCronClient(x),
)
.map((doc) => ({
name: doc.name || '',
id: String(doc.id),
schedule: doc.schedule!,
command: this.makeCommand(doc),
extra_schedules: doc.extra_schedules || [],
}));
if (isDemoEnv()) {
await writeFileWithLock(config.crontabFile, '');
return;
}
if (isDemoEnv()) {
await writeFileWithLock(config.crontabFile, '');
return;
}
// 先同步 crontab.list 与系统 crontab,确保其始终反映数据库真实状态。
// gRPC 调度注册为尽力而为:失败时不阻断文件同步,调度器重启后会重新注册。
// 这避免了因调度器短暂不可用导致 crontab.list 与数据库脱节(订阅更新误判任务已存在)。
await this.setCrontab(tabs);
try {
await cronClient.addCron(regularCrons);
} catch (error: any) {
this.logger.warn(
'[crontab] Failed to register cron job in scheduler:',
error?.message || error,
);
}
// 先同步 crontab.list 与系统 crontab,确保其始终反映数据库真实状态。
// gRPC 调度注册为尽力而为:失败时不阻断文件同步,调度器重启后会重新注册。
// 这避免了因调度器短暂不可用导致 crontab.list 与数据库脱节(订阅更新误判任务已存在)。
await this.setCrontab(tabs);
try {
await cronClient.addCron(regularCrons, requireScheduler);
} catch (error: any) {
this.logger.warn(
'[crontab] Failed to register cron job in scheduler: %s',
error?.message || error,
);
if (requireScheduler) throw error;
}
});
}
public async bootTask() {
@@ -998,13 +1147,7 @@ export default class CronService {
(x) => !x.isDisabled && this.isBootSchedule(x.schedule),
);
if (bootTasks.length > 0) {
await CrontabModel.update(
{ status: CrontabStatus.queued },
{ where: { id: bootTasks.map((t) => t.id!) } },
);
for (const task of bootTasks) {
this.runSingle(task.id!);
}
await this.run(bootTasks.map((task) => task.id!));
}
}
}
+3 -4
View File
@@ -262,10 +262,9 @@ export default class EnvService {
.map((x) => x.value)
.join('&')
.replace(/\\/g, '\\\\')}`;
js_env_string += `process.env.${key}=\`${_env_value.replace(
/\`/g,
'\\`',
)}\`;\n`;
js_env_string += `process.env[${JSON.stringify(key)}]=${JSON.stringify(
group.map((x) => x.value).join('&'),
)};\n`;
py_env_string += `os.environ['${key}']='''${_env_value.replace(
/\'/g,
"\\'",
+2 -1
View File
@@ -4,6 +4,7 @@ import { HealthService } from '../protos/health';
import { ApiService } from '../protos/api';
import { addCron } from '../schedule/addCron';
import { delCron } from '../schedule/delCron';
import { setConcurrency } from '../schedule/setConcurrency';
import { check } from '../schedule/health';
import * as Api from '../schedule/api';
import Logger from '../loaders/logger';
@@ -27,7 +28,7 @@ export class GrpcServerService {
async initialize() {
try {
this.server.addService(HealthService, { check });
this.server.addService(CronService, { addCron, delCron });
this.server.addService(CronService, { addCron, delCron, setConcurrency });
this.server.addService(ApiService, Api);
const tlsConfig = await initGrpcCerts();
+2 -4
View File
@@ -1,6 +1,6 @@
import { Service } from 'typedi';
import Logger from '../loaders/logger';
import { GrpcServerService } from './grpc';
import cronClient from '../schedule/client';
import { HttpServerService } from './http';
interface HealthStatus {
@@ -23,7 +23,6 @@ export class HealthService {
private startTime = Date.now();
constructor(
private grpcServerService: GrpcServerService,
private httpServerService: HttpServerService,
) {}
@@ -56,8 +55,7 @@ export class HealthService {
}
try {
const grpcServer = this.grpcServerService.getServer();
if (!grpcServer) {
if (!(await cronClient.readiness.check())) {
status.services.grpc = false;
status.status = 'error';
}
+7 -1
View File
@@ -39,7 +39,13 @@ export class HttpServerService {
private async tryListen(expressApp: express.Application, port: number, host: string): Promise<Server> {
return new Promise((resolve, reject) => {
const server = expressApp.listen(port, host, () => {
// There is one HTTP worker; accepting here avoids primary IPC handoff
// for every connection. Restore shared listening for custom clusters.
const server = expressApp.listen({
port,
host,
exclusive: process.env.QL_HTTP_SHARED_LISTEN !== 'true',
}, () => {
resolve(server);
});
+1 -6
View File
@@ -1,5 +1,4 @@
import { resolveFileAccess } from '../shared/fileAccess';
import path from 'path';
import { Inject, Service } from 'typedi';
import winston from 'winston';
import config from '../config';
@@ -9,10 +8,6 @@ export default class LogService {
constructor(@Inject('logger') private logger: winston.Logger) {}
public checkFilePath(filePath: string, fileName: string) {
return resolveFileAccess(
config.logPath,
[filePath || '', fileName],
config.blackFileList,
);
return resolveFileAccess(config.logPath, [filePath || '', fileName]);
}
}
+28
View File
@@ -37,6 +37,7 @@ export default class NotificationService {
['wxPusherBot', this.wxPusherBot],
['wxPusherSpt', this.wxPusherSpt],
['openiLink', this.openiLink],
['wpush', this.wpush],
]);
private title = '';
@@ -947,4 +948,31 @@ export default class NotificationService {
throw new Error(error.response ? error.response.body : error);
}
}
private async wpush() {
const { wpushApiKey, wpushChannel, wpushTopicCode } = this.params;
const url = 'https://api.wpush.cn/api/v1/send';
const json: Record<string, string> = {
apikey: `${wpushApiKey}`,
title: `${this.title}`,
content: `${this.content}`,
channel: `${wpushChannel || 'wechat'}`,
};
if (wpushTopicCode) {
json.topic_code = `${wpushTopicCode}`;
}
try {
const res = await httpClient.post(url, {
...this.gotOption,
json,
});
if (res.code === 0) {
return true;
} else {
throw new Error(JSON.stringify(res));
}
} catch (error: any) {
throw new Error(error.response ? error.response.body : error);
}
}
}
+116 -34
View File
@@ -6,27 +6,77 @@ import { v4 as uuidV4 } from 'uuid';
import sequelize, { Op } from 'sequelize';
import { shareStore } from '../shared/store';
import { t } from '../shared/i18n';
import { lock } from 'proper-lockfile';
import path from 'path';
import os from 'os';
import { writeFile, rename, rm, stat, chmod } from 'fs/promises';
import config from '../config';
@Service()
export default class OpenService {
constructor(@Inject('logger') private logger: winston.Logger) {}
// Every database mutation and full authentication-cache publication shares
// this cross-process lock. Lock a separate marker, not token.json itself.
private async withAppsLock<T>(operation: () => Promise<T>): Promise<T> {
const target = path.join(config.configPath, '.apps-state');
const release = await lock(target, {
realpath: false,
stale: 10_000,
retries: { retries: 10, factor: 2, minTimeout: 100, maxTimeout: 3000 },
});
try {
return await operation();
} finally {
await release();
}
}
public async initializeSystemApp(): Promise<void> {
await this.withAppsLock(async () => {
const doc = await AppModel.findOne({ where: { name: 'system' } });
const app = doc?.get({ plain: true });
if (!app) {
await this.insert({
name: 'system',
scopes: ['crons', 'system', 'dashboard'],
client_id: createRandomString(12, 12),
client_secret: createRandomString(24, 24),
});
} else if (!app.scopes.includes('dashboard')) {
await AppModel.update(
{ scopes: [...app.scopes, 'dashboard'] },
{ where: { id: app.id } },
);
}
await shareStore.updateApps(await this.find({}));
});
}
public async refreshApps(): Promise<void> {
await this.withAppsLock(async () => {
await shareStore.updateApps(await this.find({}));
});
}
public async findApps(): Promise<App[] | null> {
const docs = await this.find({});
return docs;
}
public async create(payload: App): Promise<App> {
const tab = { ...payload };
tab.client_id = createRandomString(12, 12);
tab.client_secret = createRandomString(24, 24);
const doc = await this.insert(tab);
const apps = await this.find({});
await shareStore.updateApps(apps);
return { ...doc, tokens: [] };
return this.withAppsLock(async () => {
const tab = { ...payload };
tab.client_id = createRandomString(12, 12);
tab.client_secret = createRandomString(24, 24);
const doc = await this.insert(tab);
const apps = await this.find({});
await shareStore.updateApps(apps);
return { ...doc, tokens: [] };
});
}
public async insert(payload: App): Promise<App> {
private async insert(payload: App): Promise<App> {
const doc = await AppModel.create(payload, { returning: true });
return doc.get({ plain: true });
}
@@ -41,10 +91,12 @@ export default class OpenService {
}
private async updateDb(payload: Partial<App>): Promise<App> {
await AppModel.update(payload, { where: { id: payload.id } });
const apps = await this.find({});
await shareStore.updateApps(apps);
return apps?.find((x) => x.id === payload.id) as App;
return this.withAppsLock(async () => {
await AppModel.update(payload, { where: { id: payload.id } });
const apps = await this.find({});
await shareStore.updateApps(apps);
return apps?.find((x) => x.id === payload.id) as App;
});
}
public async getDb(query: Record<string, any>): Promise<App> {
@@ -56,9 +108,11 @@ export default class OpenService {
}
public async remove(ids: number[]) {
await AppModel.destroy({ where: { id: ids } });
const apps = await this.find({});
await shareStore.updateApps(apps);
return this.withAppsLock(async () => {
await AppModel.destroy({ where: { id: ids } });
const apps = await this.find({});
await shareStore.updateApps(apps);
});
}
public async resetSecret(id: number): Promise<App> {
@@ -112,7 +166,15 @@ export default class OpenService {
return docs.map((x) => x.get({ plain: true }));
}
public async authToken({
public async authToken(credentials: {
client_id: string;
client_secret: string;
}): Promise<any> {
return this.withAppsLock(() => this.issueToken(credentials));
}
// Caller holds withAppsLock through database update and cache publication.
private async issueToken({
client_id,
client_secret,
}: {
@@ -156,27 +218,47 @@ export default class OpenService {
}
}
public async generateSystemToken(): Promise<{
public async generateSystemToken(renew = false): Promise<{
value: string;
expiration: number;
}> {
const apps = await shareStore.getApps();
const systemApp = apps?.find((x) => x.name === 'system');
if (!systemApp) {
throw new Error('system app not found');
}
const now = Math.round(Date.now() / 1000);
const currentToken = systemApp.tokens?.find((x) => x.expiration > now);
if (currentToken) {
return currentToken;
}
const { data } = await this.authToken({
client_id: systemApp.client_id,
client_secret: systemApp.client_secret,
return this.withAppsLock(async () => {
// Read after acquiring the lock; a concurrent issuer may have renewed.
const systemApp = await this.getDb({ name: 'system' });
const now = Math.round(Date.now() / 1000);
let appToken = systemApp.tokens
?.filter((token) => token.expiration > now)
.sort((left, right) => right.expiration - left.expiration)[0];
if (renew || !appToken) {
const { data } = await this.issueToken({
client_id: systemApp.client_id,
client_secret: systemApp.client_secret,
});
appToken = { value: data.token, expiration: data.expiration };
} else {
// Recover a prior database commit whose cache publication failed.
await shareStore.updateApps(await this.find({}));
}
const tokenFile = path.join(config.configPath, 'token.json');
const temporaryFile = `${tokenFile}.${process.pid}.tmp`;
const mode = await stat(tokenFile).then(
(file) => file.mode & 0o777,
(error: NodeJS.ErrnoException) => {
if (error.code === 'ENOENT') return 0o600;
throw error;
},
);
try {
await writeFile(temporaryFile, `${JSON.stringify(appToken)}${os.EOL}`, {
mode,
});
await chmod(temporaryFile, mode);
// Shell readers see a complete old or new document, never a partial one.
await rename(temporaryFile, tokenFile);
} finally {
await rm(temporaryFile, { force: true });
}
return appToken;
});
return {
...data,
value: data.token,
};
}
}
+112 -80
View File
@@ -1,6 +1,6 @@
import { Service, Inject } from 'typedi';
import winston from 'winston';
import nodeSchedule from 'node-schedule';
import { createCronJob, CronJob } from '../shared/cronScheduler';
import { ChildProcessWithoutNullStreams } from 'child_process';
import {
ToadScheduler,
@@ -11,6 +11,11 @@ import {
import dayjs from 'dayjs';
import taskLimit from '../shared/pLimit';
import { spawn } from 'cross-spawn';
import {
observeChildProcess,
asError,
ProcessResult,
} from '../shared/childProcess';
export interface ScheduleTaskType {
id?: number;
@@ -27,7 +32,7 @@ export interface TaskCallbacks {
startTime: dayjs.Dayjs,
) => Promise<void>;
onEnd?: (
cp: ChildProcessWithoutNullStreams,
cp: ChildProcessWithoutNullStreams | undefined,
endTime: dayjs.Dayjs,
diff: number,
) => Promise<void>;
@@ -37,7 +42,7 @@ export interface TaskCallbacks {
@Service()
export default class ScheduleService {
private scheduleStacks = new Map<string, nodeSchedule.Job>();
private scheduleStacks = new Map<string, CronJob>();
private intervalSchedule = new ToadScheduler();
@@ -63,73 +68,85 @@ export default class ScheduleService {
) {
const { runOrigin, ...others } = params;
return taskLimit[this.taskLimitMap[runOrigin]](others, () => {
return new Promise(async (resolve, reject) => {
this.logger.info(
`[panel][开始执行任务] 参数: ${JSON.stringify({
...others,
command,
})}`,
);
try {
const startTime = dayjs();
await callbacks.onBefore?.(startTime);
const cp = spawn(command, { shell: '/bin/bash' });
callbacks.onStart?.(cp, startTime);
completionTime === 'start' && resolve(cp.pid);
cp.stdout.on('data', async (data) => {
await callbacks.onLog?.(data.toString());
});
cp.stderr.on('data', async (data) => {
this.logger.info(
'[panel][执行任务失败] 命令: %s, 错误信息: %j',
command,
data.toString(),
);
await callbacks.onError?.(data.toString());
});
cp.on('error', async (err) => {
this.logger.error(
'[panel][创建任务失败] 命令: %s, 错误信息: %j',
command,
err,
);
await callbacks.onError?.(JSON.stringify(err));
});
cp.on('exit', async (code) => {
this.logger.info(
'[panel][执行任务结束] 参数: %s, 退出码: %j',
JSON.stringify({
...others,
command,
}),
code,
);
const endTime = dayjs();
await callbacks.onEnd?.(
cp,
endTime,
endTime.diff(startTime, 'seconds'),
);
resolve({ ...others, pid: cp.pid, code });
});
} catch (error) {
this.logger.error(
'[panel][执行任务失败] 命令: %s, 错误信息: %j',
command,
error,
);
await callbacks.onError?.(JSON.stringify(error));
}
});
let resolveStart!: (pid: number | undefined) => void;
let rejectStart!: (error: Error) => void;
const startResult = new Promise<number | undefined>((resolve, reject) => {
resolveStart = resolve;
rejectStart = reject;
});
// Most scheduled callers only observe completion (or intentionally detach).
void startResult.catch(() => {});
const completion = taskLimit[this.taskLimitMap[runOrigin]](
others,
async () => {
const startTime = dayjs();
let cp: ChildProcessWithoutNullStreams | undefined;
let result: ProcessResult = { code: null, signal: null };
try {
this.logger.info('[panel][开始执行任务] 任务ID: %s', others.id);
await callbacks.onBefore?.(startTime);
cp = spawn(command, { shell: '/bin/bash' });
const child = cp;
const observed = observeChildProcess(child, {
onStart: async () => {
await callbacks.onStart?.(child, startTime);
},
onStdout: callbacks.onLog,
onStderr: callbacks.onError,
});
observed.started.then(resolveStart, rejectStart);
result = await observed.completed;
} catch (error) {
result.error = asError(error);
rejectStart(result.error);
}
if (result.error) {
this.logger.error(
'[panel][执行任务失败] 任务ID: %s, 错误: %s',
others.id,
result.error.message,
);
try {
await callbacks.onError?.(result.error.message);
} catch (error) {
this.logger.error(
'[panel][任务错误回调失败] %s',
asError(error).message,
);
}
}
// Cleanup also runs after setup/spawn failure, and only after both pipes drain.
const endTime = dayjs();
try {
await callbacks.onEnd?.(
cp,
endTime,
endTime.diff(startTime, 'seconds'),
);
} catch (error) {
result.error ??= asError(error);
this.logger.error(
'[panel][任务结束回调失败] %s',
asError(error).message,
);
}
this.logger.info(
'[panel][执行任务结束] 任务ID: %s, 退出码: %j',
others.id,
result.code,
);
return { ...others, pid: cp?.pid, ...result };
},
).catch((error) => {
// Queue/setup failures must not become unhandled rejections in detached callers.
rejectStart(asError(error));
this.logger.error('[panel][任务队列失败] %s', asError(error).message);
return { ...others, code: null, signal: null, error: asError(error) };
});
// Returning a PID must not release the execution slot while the process runs.
return completionTime === 'start' ? startResult : completion;
}
async createCronTask(
@@ -146,18 +163,33 @@ export default class ScheduleService {
command,
);
this.scheduleStacks.set(
_id,
nodeSchedule.scheduleJob(_id, schedule, async () => {
this.runTask(command, callbacks, {
name,
if (schedule) {
try {
const job = createCronJob(
schedule,
command,
id: _id,
runOrigin,
});
}),
);
() =>
this.runTask(command, callbacks, {
name,
schedule,
command,
id: _id,
runOrigin,
}),
{ name: `${_id}: ${name || ''}`, logger: this.logger, start: false },
);
this.scheduleStacks.get(_id)?.cancel();
this.scheduleStacks.set(_id, job);
job.start();
} catch (error) {
// A persisted invalid subscription must not become an unhandled startup rejection.
this.logger.warn(
'[panel][跳过无效定时任务] 任务ID: %s, cron: %s, 错误: %s',
_id,
schedule,
asError(error).message,
);
}
}
if (runImmediately) {
this.runTask(command, callbacks, {
+4 -8
View File
@@ -58,19 +58,15 @@ export default class ScriptService {
taskLimit.removeQueuedCron(relativePath.replace(/ /g, '-'));
pid = (await getPid(`${TASK_COMMAND} ${relativePath} now`)) as number;
}
try {
await killTask(pid);
} catch (error) {}
if (pid) {
await killTask(pid, true);
}
return { code: 200 };
}
public checkFilePath(filePath: string, fileName: string) {
return resolveFileAccess(
config.scriptPath,
[filePath || '', fileName],
config.blackFileList,
);
return resolveFileAccess(config.scriptPath, [filePath || '', fileName]);
}
public async getFile(filePath: string, fileName: string) {
+12
View File
@@ -0,0 +1,12 @@
import path from 'path';
import { Service } from 'typedi';
import config from '../config';
import { ScriptHistory } from '../shared/scriptHistory';
/** Panel configuration and lifecycle; the storage core stays independently testable. */
@Service()
export default class ScriptHistoryService extends ScriptHistory {
constructor() {
super(config.scriptPath, path.join(config.dataPath, 'script-history'));
}
}
+43 -23
View File
@@ -8,10 +8,22 @@ import { formatUrl } from '../config/subscription';
import config from '../config';
import { fileExist, rmPath } from '../config/util';
import { writeFileWithLock } from '../shared/utils';
import {
getSubscriptionSshAlias,
resolveSubscriptionPath,
} from '../shared/subscriptionPath';
@Service()
export default class SshKeyService {
private homedir = os.homedir();
// OpenSSH uses the effective user's passwd entry, not the HOME override.
private get homedir(): string {
try {
return os.userInfo().homedir;
} catch {
// Custom container UIDs may have no passwd entry. Keep startup working.
return os.homedir();
}
}
private sshPath = config.sshdPath;
private sshConfigFilePath = path.resolve(this.homedir, '.ssh', 'config');
private sshConfigHeader = `Include ${path.join(this.sshPath, '*.config')}`;
@@ -21,6 +33,10 @@ export default class SshKeyService {
}
private async initSshConfigFile() {
await fs.mkdir(path.dirname(this.sshConfigFilePath), {
recursive: true,
mode: 0o700,
});
let config = '';
const _exist = await fileExist(this.sshConfigFilePath);
if (_exist) {
@@ -42,7 +58,7 @@ export default class SshKeyService {
key: string,
): Promise<void> {
try {
const filePath = path.join(this.sshPath, alias);
const filePath = resolveSubscriptionPath(this.sshPath, alias);
try {
await rmPath(filePath);
} catch { }
@@ -54,7 +70,7 @@ export default class SshKeyService {
private async removePrivateKeyFile(alias: string): Promise<void> {
try {
const filePath = path.join(this.sshPath, alias);
const filePath = resolveSubscriptionPath(this.sshPath, alias);
await rmPath(filePath);
} catch (error) {
this.logger.error('删除私钥文件失败', error);
@@ -66,18 +82,19 @@ export default class SshKeyService {
host: string,
proxy?: string,
) {
if (!/^[a-zA-Z0-9.-]+$/.test(host) ||
(proxy && !/^[a-zA-Z0-9.:[\]-]+$/.test(proxy))) {
throw Object.assign(new Error('Invalid SSH host or proxy'), { status: 400 });
}
if (host === 'github.com') {
host = `ssh.github.com\n Port 443\n HostkeyAlgorithms +ssh-rsa`;
}
const proxyStr = proxy
? ` ProxyCommand nc -v -x ${proxy} %h %p 2>/dev/null\n`
: '';
const config = `Host ${alias}\n Hostname ${host}\n IdentityFile ${path.join(
this.sshPath,
alias,
)}\n StrictHostKeyChecking no\n${proxyStr}`;
const config = `Host ${alias}\n Hostname ${host}\n IdentityFile ${resolveSubscriptionPath(this.sshPath, alias)}\n StrictHostKeyChecking no\n${proxyStr}`;
await writeFileWithLock(
`${path.join(this.sshPath, `${alias}.config`)}`,
`${resolveSubscriptionPath(this.sshPath, alias, '.config')}`,
config,
{
encoding: 'utf8',
@@ -88,7 +105,7 @@ export default class SshKeyService {
private async removeSshConfig(alias: string) {
try {
const filePath = path.join(this.sshPath, `${alias}.config`);
const filePath = resolveSubscriptionPath(this.sshPath, alias, '.config');
await rmPath(filePath);
} catch (error) {
this.logger.error(`删除ssh配置文件${alias}失败`, error);
@@ -101,6 +118,7 @@ export default class SshKeyService {
host: string,
proxy?: string,
): Promise<void> {
alias = getSubscriptionSshAlias(alias);
await this.generatePrivateKeyFile(alias, key);
await this.generateSingleSshConfig(alias, host, proxy);
}
@@ -110,6 +128,7 @@ export default class SshKeyService {
host: string,
proxy?: string,
): Promise<void> {
alias = getSubscriptionSshAlias(alias);
await this.removePrivateKeyFile(alias);
await this.removeSshConfig(alias);
}
@@ -117,15 +136,19 @@ export default class SshKeyService {
public async setSshConfig(docs: Subscription[]) {
for (const doc of docs) {
if (doc.type === 'private-repo' && doc.pull_type === 'ssh-key') {
const { alias, proxy } = doc;
const { host } = formatUrl(doc);
await this.removePrivateKeyFile(alias);
await this.removeSshConfig(alias);
await this.generatePrivateKeyFile(
alias,
(doc.pull_option as any).private_key,
);
await this.generateSingleSshConfig(alias, host, proxy);
try {
const alias = getSubscriptionSshAlias(doc.alias);
const { host } = formatUrl(doc);
await this.removePrivateKeyFile(alias);
await this.removeSshConfig(alias);
await this.generatePrivateKeyFile(
alias,
(doc.pull_option as any).private_key,
);
await this.generateSingleSshConfig(alias, host, doc.proxy);
} catch (error) {
this.logger.warn('跳过订阅 %s 的无效 SSH 配置: %s', doc.id, error);
}
}
}
}
@@ -144,12 +167,9 @@ export default class SshKeyService {
private async generateGlobalSshConfig(alias: string) {
// Create a config that matches all hosts, making this key globally available
const config = `Host *\n IdentityFile ${path.join(
this.sshPath,
alias,
)}\n StrictHostKeyChecking no\n`;
const config = `Host *\n IdentityFile ${resolveSubscriptionPath(this.sshPath, alias)}\n StrictHostKeyChecking no\n`;
await writeFileWithLock(
`${path.join(this.sshPath, `${alias}.config`)}`,
`${resolveSubscriptionPath(this.sshPath, alias, '.config')}`,
config,
{
encoding: 'utf8',
+164 -81
View File
@@ -20,7 +20,7 @@ import {
} from '../config/util';
import fs from 'fs/promises';
import { FindOptions, Op } from 'sequelize';
import path, { join } from 'path';
import path from 'path';
import ScheduleService, { TaskCallbacks } from './schedule';
import { SimpleIntervalSchedule } from 'toad-scheduler';
import SockService from './sock';
@@ -30,6 +30,14 @@ import dayjs from 'dayjs';
import { LOG_END_SYMBOL } from '../config/const';
import { formatCommand, formatUrl } from '../config/subscription';
import { CrontabModel } from '../data/cron';
import { assertSubscriptionAlias } from '../shared/subscriptionPath';
import { resolveLogPath } from '../shared/logPath';
import {
getSubscriptionStorageCandidates,
isSubscriptionStorageName,
resolveSubscriptionStoragePath,
} from '../shared/subscriptionStorage';
import { withSubscriptionMutation } from '../shared/subscriptionMutationLock';
import CrontabService from './cron';
import taskLimit from '../shared/pLimit';
import { logStreamManager } from '../shared/logStreamManager';
@@ -88,9 +96,11 @@ export default class SubscriptionService {
needCreate = true,
runImmediately = false,
) {
const { url } = formatUrl(doc);
doc.command = formatCommand(doc, url as string);
// Cancelling by ID must also work for invalid historical subscriptions.
if (needCreate) {
const { url } = formatUrl(doc);
doc.command = formatCommand(doc, url as string);
}
if (doc.schedule_type === 'crontab') {
this.scheduleService.cancelCronTask(doc as any);
@@ -159,48 +169,54 @@ export default class SubscriptionService {
);
},
onEnd: async (cp, endTime, diff) => {
const sub = await this.getDb({ id: doc.id });
const absolutePath = await handleLogPath(sub.log_path as string);
// 执行 sub_after
let afterStr = '';
let absolutePath: string | undefined;
try {
if (sub.sub_after) {
await logStreamManager.write(absolutePath, `\n\n## ${t('执行after命令...')}\n\n`);
afterStr = await promiseExec(sub.sub_after);
const sub = await this.getDb({ id: doc.id });
absolutePath = await handleLogPath(sub.log_path as string);
// 执行 sub_after
let afterStr = '';
try {
if (sub.sub_after) {
await logStreamManager.write(
absolutePath,
`\n\n## ${t('执行after命令...')}\n\n`,
);
afterStr = await promiseExec(sub.sub_after);
}
} catch (error: any) {
afterStr =
(error.stderr && error.stderr.toString()) || JSON.stringify(error);
}
if (afterStr) {
await logStreamManager.write(absolutePath, `${afterStr}\n`);
}
await logStreamManager.write(
absolutePath,
'\n' +
tf(
'## 执行结束... %s 耗时 %s 秒',
endTime.format('YYYY-MM-DD HH:mm:ss'),
String(diff),
) +
LOG_END_SYMBOL,
);
} finally {
try {
if (absolutePath) await logStreamManager.closeStream(absolutePath);
} finally {
await SubscriptionModel.update(
{ status: SubscriptionStatus.idle, pid: null } as any,
{ where: { id: doc.id } },
);
this.sockService.sendMessage({
type: 'runSubscriptionEnd',
message: t('订阅执行完成'),
references: [doc.id as number],
});
}
} catch (error: any) {
afterStr =
(error.stderr && error.stderr.toString()) || JSON.stringify(error);
}
if (afterStr) {
await logStreamManager.write(absolutePath, `${afterStr}\n`);
}
await logStreamManager.write(
absolutePath,
'\n' +
tf(
'## 执行结束... %s 耗时 %s 秒',
endTime.format('YYYY-MM-DD HH:mm:ss'),
String(diff),
) +
LOG_END_SYMBOL,
);
// Close the stream after task completion
await logStreamManager.closeStream(absolutePath);
await SubscriptionModel.update(
{ status: SubscriptionStatus.idle, pid: undefined },
{ where: { id: sub.id } },
);
this.sockService.sendMessage({
type: 'runSubscriptionEnd',
message: t('订阅执行完成'),
references: [doc.id as number],
});
},
onError: async (message: string) => {
const sub = await this.getDb({ id: doc.id });
@@ -216,11 +232,14 @@ export default class SubscriptionService {
}
public async create(payload: Subscription): Promise<Subscription> {
const tab = new Subscription(payload);
const doc = await this.insert(tab);
await this.handleTask(doc.get({ plain: true }));
await this.setSshConfig();
return doc;
assertSubscriptionAlias(payload.alias);
return withSubscriptionMutation(async () => {
const tab = new Subscription(payload);
const doc = await this.insert(tab);
await this.handleTask(doc.get({ plain: true }));
await this.setSshConfig();
return doc;
});
}
public async insert(payload: Subscription): Promise<SubscriptionInstance> {
@@ -228,12 +247,27 @@ export default class SubscriptionService {
}
public async update(payload: Subscription): Promise<Subscription> {
const doc = await this.getDb({ id: payload.id });
const tab = new Subscription({ ...doc, ...payload });
const newDoc = await this.updateDb(tab);
await this.handleTask(newDoc, !newDoc.is_disabled);
await this.setSshConfig();
return newDoc;
assertSubscriptionAlias(payload.alias);
return withSubscriptionMutation(async () => {
const doc = await this.getDb({ id: payload.id });
const tab = new Subscription({ ...doc, ...payload });
const newDoc = await this.updateDb(tab);
await this.handleTask(newDoc, !newDoc.is_disabled);
await this.removeSshConfigForSubscription(doc);
await this.setSshConfig();
return newDoc;
});
}
private async removeSshConfigForSubscription(doc: Subscription) {
if (doc.type !== 'private-repo' || doc.pull_type !== 'ssh-key') return;
// Invalid historical rows must remain deletable without deriving unsafe paths.
try {
assertSubscriptionAlias(doc.alias);
} catch {
return;
}
await this.sshKeyService.removeSSHKey(doc.alias, '');
}
public async updateDb(payload: Subscription): Promise<Subscription> {
@@ -256,6 +290,7 @@ export default class SubscriptionService {
last_running_time: number;
last_execution_time: number;
}) {
if (log_path) resolveLogPath(config.logPath, log_path);
const options: any = {
status,
pid,
@@ -273,25 +308,71 @@ export default class SubscriptionService {
}
public async remove(ids: number[], query: { force?: boolean }) {
const docs = await SubscriptionModel.findAll({ where: { id: ids } });
for (const doc of docs) {
await this.handleTask(doc.get({ plain: true }), false);
}
await SubscriptionModel.destroy({ where: { id: ids } });
await this.setSshConfig();
if (query?.force === true) {
const crons = await CrontabModel.findAll({ where: { sub_id: ids } });
if (crons?.length) {
await this.crontabService.remove(crons.map((x) => x.id!));
return withSubscriptionMutation(async () => {
const docs = await SubscriptionModel.findAll({ where: { id: ids } });
const storagePaths: string[] = [];
const retainedPaths: string[] = [];
const collectPaths = (
doc: Subscription,
targets: string[],
deleting = false,
) => {
for (const names of getSubscriptionStorageCandidates(doc)) {
// A URL accepted by one engine can produce an invalid name in another.
// Validate each candidate independently, retaining the safe candidate.
if (!Object.values(names).every(isSubscriptionStorageName)) continue;
const resolve = deleting ? resolveSubscriptionStoragePath : path.resolve;
const scriptPath = resolve(config.scriptPath, names.script);
if (scriptPath) targets.push(scriptPath);
if (names.repo) {
const repoPath = resolve(config.repoPath, names.repo);
if (repoPath) targets.push(repoPath);
}
if (names.raw) {
const rawPath = resolve(path.join(config.dataPath, 'raw'), names.raw);
if (rawPath) targets.push(rawPath);
}
}
};
// Validate all deletion targets before changing rows or removing any files.
if (query?.force === true) {
for (const doc of docs) {
assertSubscriptionAlias(doc.alias);
resolveSubscriptionStoragePath(config.scriptPath, doc.alias);
resolveSubscriptionStoragePath(config.repoPath, doc.alias);
collectPaths(doc, storagePaths, true);
}
const remaining = await SubscriptionModel.findAll({ where: {} });
for (const doc of remaining) {
if (!ids.includes(doc.id!)) collectPaths(doc, retainedPaths);
}
}
for (const doc of docs) {
const filePath = join(config.scriptPath, doc.alias);
const repoPath = join(config.repoPath, doc.alias);
await rmPath(filePath);
await rmPath(repoPath);
await this.handleTask(doc.get({ plain: true }), false);
}
}
await SubscriptionModel.destroy({ where: { id: ids } });
for (const doc of docs) {
await this.removeSshConfigForSubscription(doc);
}
await this.setSshConfig();
if (query?.force === true) {
const crons = await CrontabModel.findAll({ where: { sub_id: ids } });
if (crons?.length) {
await this.crontabService.remove(crons.map((x) => x.id!));
}
for (const storagePath of new Set(storagePaths)) {
// rm is recursive: an ancestor must also be retained if a surviving
// subscription uses a nested branch, and vice versa.
const shared = retainedPaths.some((retained) =>
retained === storagePath ||
retained.startsWith(`${storagePath}${path.sep}`) ||
storagePath.startsWith(`${retained}${path.sep}`),
);
if (!shared) await rmPath(storagePath);
}
}
});
}
public async getDb(
@@ -316,20 +397,22 @@ export default class SubscriptionService {
public async stop(ids: number[]) {
const docs = await SubscriptionModel.findAll({ where: { id: ids } });
let failure: unknown;
for (const doc of docs) {
if (doc.pid) {
try {
await killTask(doc.pid);
} catch (error) {
this.logger.error(error);
try {
if (doc.pid) {
await killTask(doc.pid, true);
}
await SubscriptionModel.update(
{ status: SubscriptionStatus.idle, pid: null } as any,
{ where: { id: doc.id, pid: doc.pid ?? null } },
);
} catch (error) {
this.logger.error(error);
failure ??= error;
}
}
await SubscriptionModel.update(
{ status: SubscriptionStatus.idle, pid: undefined },
{ where: { id: ids } },
);
if (failure) throw failure;
}
private async runSingle(subscriptionId: number) {
@@ -391,7 +474,7 @@ export default class SubscriptionService {
if (doc.log_path) {
const relativeDir = path.dirname(`${doc.log_path}`);
const dir = path.resolve(config.logPath, relativeDir);
const dir = path.dirname(resolveLogPath(config.logPath, doc.log_path));
const _exist = await fileExist(dir);
if (_exist) {
let files = await fs.readdir(dir);
+52 -22
View File
@@ -1,4 +1,7 @@
import { spawn } from 'cross-spawn';
import { execFile } from 'child_process';
import { promisify } from 'util';
import { shellQuote } from '../shared/shellQuote';
import { Response } from 'express';
import fs from 'fs';
import { Agent, request } from 'undici';
@@ -31,6 +34,8 @@ import {
SystemModelInfo,
} from '../data/system';
import taskLimit from '../shared/pLimit';
import cronClient from '../schedule/client';
import { withSchedulerMutation } from '../shared/schedulerMutationLock';
import NotificationService from './notify';
import ScheduleService, { TaskCallbacks } from './schedule';
import SockService from './sock';
@@ -119,15 +124,29 @@ export default class SystemService {
}
public async updateCronConcurrency(info: SystemModelInfo) {
const oDoc = await this.getSystemConfig();
await this.updateAuthDb({
...oDoc,
info: { ...oDoc.info, ...info },
});
if (info.cronConcurrency) {
await taskLimit.setCustomLimit(info.cronConcurrency);
const limit = info.cronConcurrency ?? 0;
if (!Number.isInteger(limit) || limit < 0 || limit > 2147483647) {
throw Object.assign(new Error(t('参数错误')), { status: 400 });
}
return { code: 200, data: info };
return withSchedulerMutation(async () => {
const oDoc = await this.getSystemConfig();
await this.updateAuthDb({
...oDoc,
info: { ...oDoc.info, ...info, cronConcurrency: limit },
});
await taskLimit.setCustomLimit(limit || Math.max(os.cpus().length, 4));
await cronClient.setConcurrency(limit);
return { code: 200, data: info };
});
}
public async restoreCronConcurrency() {
await withSchedulerMutation(async () => {
const doc = await this.getSystemConfig();
const limit = doc.info?.cronConcurrency ?? 0;
await taskLimit.setCustomLimit(limit || Math.max(os.cpus().length, 4));
await cronClient.setConcurrency(limit);
});
}
public async updateDependenceProxy(info: SystemModelInfo) {
@@ -139,7 +158,7 @@ export default class SystemService {
if (info.dependenceProxy) {
await fs.promises.writeFile(
config.dependenceProxyFile,
`export http_proxy="${info.dependenceProxy}"\nexport https_proxy="${info.dependenceProxy}"`,
`export http_proxy=${shellQuote(info.dependenceProxy)}\nexport https_proxy=${shellQuote(info.dependenceProxy)}`,
);
} else {
await fs.promises.rm(config.dependenceProxyFile);
@@ -155,7 +174,7 @@ export default class SystemService {
});
let cmd = 'pnpm config delete registry';
if (info.nodeMirror) {
cmd = `pnpm config set registry ${info.nodeMirror}`;
cmd = `pnpm config set registry ${shellQuote(info.nodeMirror)}`;
}
let command = `cd && ${cmd}`;
const docs = await DependenceModel.findAll({
@@ -204,7 +223,7 @@ export default class SystemService {
});
let cmd = 'pip config unset global.index-url';
if (info.pythonMirror) {
cmd = `pip3 config set global.index-url ${info.pythonMirror}`;
cmd = `pip3 config set global.index-url ${shellQuote(info.pythonMirror)}`;
}
await promiseExec(cmd);
return { code: 200, data: info };
@@ -220,7 +239,6 @@ export default class SystemService {
return;
}
const command = await updateLinuxMirrorFile(info.linuxMirror || '');
let hasError = false;
this.scheduleService.runTask(
command,
{
@@ -228,14 +246,15 @@ export default class SystemService {
res?.setHeader('QL-Task-Pid', `${cp.pid}`);
res?.end();
},
onEnd: async () => {
onEnd: async (cp) => {
this.sockService.sendMessage({
type: 'updateLinuxMirror',
message: 'update linux mirror end',
status: 'completed',
});
onEnd?.();
if (!hasError) {
// Tools such as sudo and apt may write warnings to stderr on success.
if (cp?.exitCode === 0 && cp.signalCode === null) {
await this.updateAuthDb({
...oDoc,
info: { ...oDoc.info, ...info },
@@ -243,7 +262,6 @@ export default class SystemService {
}
},
onError: async (message: string) => {
hasError = true;
this.sockService.sendMessage({ type: 'updateLinuxMirror', message });
},
onLog: async (message: string) => {
@@ -357,7 +375,10 @@ export default class SystemService {
}
public async reloadSystem(target?: 'system' | 'data') {
const cmd = `real_time=true ql reload ${target || ''}`;
if (target && !['system', 'data'].includes(target)) {
return { code: 400, message: t('参数错误') };
}
const cmd = `real_time=true ql reload ${shellQuote(target || '')}`;
const cp = spawn(cmd, {
shell: '/bin/bash',
detached: true,
@@ -438,11 +459,20 @@ export default class SystemService {
if (type && type.length) {
dataDirs = dataDirs.concat(type.filter((x) => x !== 'base'));
}
const dataPaths = dataDirs.map((dir) => `data/${dir}`);
await promiseExec(
`cd ${config.dataPath} && cd ../ && tar -zcvf ${config.dataTgzFile
} ${dataPaths.join(' ')}`,
);
const allowed = new Set(['db', 'upload', 'config', 'scripts', 'log', 'deps',
'syslog', 'dep_cache', 'raw', 'repo', 'ssh.d', 'script-history']);
if (dataDirs.some((dir) => !allowed.has(dir))) {
return res.status(400).send({ code: 400, message: t('参数错误') });
}
if (dataDirs.includes('scripts')) dataDirs.push('script-history');
// History is created lazily; old/fresh installations must still export.
if (!fs.existsSync(path.join(config.dataPath, 'script-history'))) {
dataDirs = dataDirs.filter((dir) => dir !== 'script-history');
}
const dataPaths = [...new Set(dataDirs)].map((dir) => `data/${dir}`);
await promisify(execFile)('tar', ['-zcf', config.dataTgzFile, '--', ...dataPaths], {
cwd: path.dirname(config.dataPath),
});
res.download(config.dataTgzFile);
} catch (error: any) {
return res.send({ code: 400, message: error.message });
@@ -565,7 +595,7 @@ export default class SystemService {
try {
await fs.promises.writeFile(
config.langEnvFile,
`export QL_LANG='${lang}'\n`,
`export QL_LANG=${shellQuote(lang)}\n`,
);
} catch (error) {
this.logger.error(`Failed to write lang_env.sh: ${error}`);
+36 -12
View File
@@ -33,9 +33,14 @@ import {
verifyPassword,
} from '../shared/password';
import { serializeAuthMutation } from '../shared/authMutation';
import { LoginThrottle } from '../shared/loginThrottle';
import UserManagementService from './userManagement';
@Service()
export default class UserService {
private loginThrottle = new LoginThrottle();
@Inject((type) => UserManagementService)
private userManagementService!: UserManagementService;
@Inject((type) => NotificationService)
private notificationService!: NotificationService;
@@ -106,11 +111,8 @@ export default class UserService {
return { code: 403, message: t('该 IP 已被列入黑名单') };
}
const retriesTime = Math.pow(3, retries) * 1000;
if (retries > 2 && timestamp - lastlogon < retriesTime) {
const waitTime = Math.ceil(
(retriesTime - (timestamp - lastlogon)) / 1000,
);
const waitTime = this.loginThrottle.retryAfter(ip, timestamp);
if (waitTime > 0) {
return {
code: 410,
message: tf('失败次数过多,请%s秒后重试', waitTime),
@@ -118,6 +120,24 @@ export default class UserService {
};
}
if (username !== cUsername) {
const account = await this.userManagementService.authenticate(
username,
password,
);
if (account) {
this.loginThrottle.reset(ip);
const token = await this.userManagementService.issueToken(
account,
req.platform,
);
return {
code: 200,
data: { token, username: account.username, role: account.role },
};
}
}
const passwordMatches =
username === cUsername && (await verifyPassword(password, cPassword));
@@ -133,6 +153,7 @@ export default class UserService {
}
if (passwordMatches) {
this.loginThrottle.reset(ip);
const data = createRandomString(50, 100);
const expiration = twoFactorActivated ? '60d' : '20d';
let token = jwt.sign({ data }, config.jwt.secret, {
@@ -206,8 +227,9 @@ export default class UserService {
},
};
} else {
const failures = this.loginThrottle.fail(ip, timestamp);
await this.updateAuthInfo(content, {
retries: retries + 1,
retries: failures,
lastlogon: timestamp,
lastip: ip,
lastaddr: address,
@@ -239,8 +261,8 @@ export default class UserService {
},
});
this.getLoginLog();
if (retries > 2) {
const waitTime = Math.round(Math.pow(3, retries + 1));
if (failures > 3) {
const waitTime = this.loginThrottle.retryAfter(ip, timestamp);
return {
code: 410,
message: tf('失败次数过多,请%s秒后重试', waitTime),
@@ -439,8 +461,8 @@ export default class UserService {
const authInfo = await this.getAuthInfo();
const { isTwoFactorChecking, twoFactorSecret } = authInfo;
const now = Date.now();
const retries = authInfo.retries || 0;
if (retries > 2 && now - authInfo.lastlogon < Math.pow(3, retries) * 1000) {
const source = getClientIp(req);
if (this.loginThrottle.retryAfter(source, now) > 0) {
return { code: 410, message: t('失败次数过多,请稍后重试') };
}
if (
@@ -472,9 +494,8 @@ export default class UserService {
.join(' ');
}
await this.updateAuthInfo(authInfo, {
retries: retries + 1,
retries: this.loginThrottle.fail(source, now),
lastlogon: now,
isTwoFactorChecking: retries + 1 < 5,
lastip: ip,
lastaddr: address,
platform: req.platform,
@@ -690,5 +711,8 @@ export default class UserService {
});
}
await this.updateAuthInfo(authInfo, payload);
if (retries === 0 || password !== undefined || username !== undefined) {
this.loginThrottle.clear();
}
}
}
+160
View File
@@ -0,0 +1,160 @@
import { Service, Inject } from 'typedi';
import winston from 'winston';
import { User, UserModel, UserStatus } from '../data/user';
import { Op } from 'sequelize';
import { createHash, randomBytes } from 'crypto';
import jwt from 'jsonwebtoken';
import config from '../config';
import {
hashPassword,
verifyPassword,
isPasswordHash,
} from '../shared/password';
import { serializeAuthMutation } from '../shared/authMutation';
import { shareStore } from '../shared/store';
type PublicUser = Omit<User, 'password' | 'sessions'>;
function publicUser(user: User): PublicUser {
const { password, sessions, ...result } = user;
return result;
}
function digest(token: string) {
return createHash('sha256').update(token).digest('hex');
}
@Service()
export default class UserManagementService {
constructor(@Inject('logger') private logger: winston.Logger) {}
public async list(searchText?: string): Promise<PublicUser[]> {
const where = searchText
? { username: { [Op.like]: `%${searchText}%` } }
: {};
return (await UserModel.findAll({ where })).map((x) =>
publicUser(x.get({ plain: true })),
);
}
public async get(id: number): Promise<PublicUser> {
const doc = await UserModel.findByPk(id);
if (!doc) throw Object.assign(new Error('用户不存在'), { status: 404 });
return publicUser(doc.get({ plain: true }));
}
public async getByUsername(username: string): Promise<User | null> {
return (
(await UserModel.findOne({ where: { username } }))?.get({
plain: true,
}) || null
);
}
@serializeAuthMutation
public async create(payload: User): Promise<PublicUser> {
const auth = await shareStore.getAuthInfo();
if (
auth?.username === payload.username ||
(await this.getByUsername(payload.username))
)
throw Object.assign(new Error('用户名已存在'), { status: 400 });
if (!payload.password || payload.password.length < 6)
throw Object.assign(new Error('密码长度至少为6位'), { status: 400 });
const user = new User({
...payload,
password: await hashPassword(payload.password),
sessions: [],
});
return publicUser((await UserModel.create(user)).get({ plain: true }));
}
@serializeAuthMutation
public async update(
payload: Partial<User> & { id: number },
): Promise<PublicUser> {
const existing = await this.get(payload.id);
if (payload.username && payload.username !== existing.username) {
const auth = await shareStore.getAuthInfo();
if (
auth?.username === payload.username ||
(await this.getByUsername(payload.username))
)
throw Object.assign(new Error('用户名已存在'), { status: 400 });
}
const changes: Partial<User> = {};
for (const key of ['username', 'role', 'status'] as const)
if (payload[key] !== undefined) (changes as any)[key] = payload[key];
if (payload.password) {
if (payload.password.length < 6)
throw Object.assign(new Error('密码长度至少为6位'), { status: 400 });
changes.password = await hashPassword(payload.password);
}
if (Object.keys(changes).length) changes.sessions = [];
await UserModel.update(changes, { where: { id: payload.id } });
return this.get(payload.id);
}
@serializeAuthMutation
public async delete(ids: number[]): Promise<number> {
return UserModel.destroy({ where: { id: ids } });
}
public async authenticate(
username: string,
password: string,
): Promise<User | null> {
const user = await this.getByUsername(username);
if (
!user ||
user.status !== UserStatus.active ||
!isPasswordHash(user.password)
)
return null;
return (await verifyPassword(password, user.password)) ? user : null;
}
// Caller holds the shared account mutation queue during authentication and issuance.
public async issueToken(user: User, platform: string): Promise<string> {
const token = jwt.sign(
{ userId: user.id, nonce: randomBytes(32).toString('hex') },
config.jwt.secret,
{ expiresIn: config.jwt.expiresIn || '20d', algorithm: 'HS384' },
);
const sessions = [
...(user.sessions || []).filter((x) => x.platform !== platform),
{ digest: digest(token), platform },
].slice(-10);
await UserModel.update({ sessions }, { where: { id: user.id } });
return token;
}
public async validateToken(
token: string,
platform: string,
): Promise<PublicUser | null> {
try {
const payload = jwt.verify(token, config.jwt.secret, {
algorithms: ['HS384'],
});
if (
typeof payload === 'string' ||
!Number.isSafeInteger(payload.userId) ||
payload.userId <= 0
)
return null;
const user = (await UserModel.findByPk(payload.userId))?.get({
plain: true,
});
if (
!user ||
user.status !== UserStatus.active ||
!user.sessions?.some(
(x) => x.digest === digest(token) && x.platform === platform,
)
)
return null;
return publicUser(user);
} catch {
return null;
}
}
@serializeAuthMutation
public async revokeToken(id: number, token: string): Promise<void> {
const user = await UserModel.findByPk(id);
if (!user) return;
await user.update({
sessions: (user.sessions || []).filter((x) => x.digest !== digest(token)),
});
}
}
+78
View File
@@ -0,0 +1,78 @@
import { ChildProcessWithoutNullStreams } from 'child_process';
import { Readable } from 'stream';
export interface ProcessResult {
code: number | null;
signal: NodeJS.Signals | null;
error?: Error;
}
export function asError(error: unknown): Error {
return error instanceof Error ? error : new Error(String(error));
}
/** Attach immediately after spawn, before awaiting database or user callbacks. */
export function observeChildProcess(
child: ChildProcessWithoutNullStreams,
callbacks: {
onStart?: () => Promise<void>;
onStdout?: (message: string) => Promise<void>;
onStderr?: (message: string) => Promise<void>;
} = {},
) {
let failure: Error | undefined;
const recordError = (error: unknown) => {
failure ??= asError(error);
};
const spawned = new Promise<void>((resolve, reject) => {
child.once('spawn', resolve);
// Keep the listener through close: errors can occur after a successful spawn.
child.on('error', (error) => {
recordError(error);
reject(error);
});
});
const closed = new Promise<ProcessResult>((resolve) => {
child.once('close', (code, signal) => resolve({ code, signal }));
});
const started = spawned.then(async () => {
await callbacks.onStart?.();
return child.pid;
});
// The caller can ask only for completion, without an unhandled start rejection.
const ready = started.catch(recordError);
const consume = async (
stream: Readable,
callback?: (message: string) => Promise<void>,
) => {
// StringDecoder in Readable preserves UTF-8 characters split across chunks.
stream.setEncoding('utf8');
let callbackFailed = false;
try {
for await (const chunk of stream) {
await ready;
if (!callbackFailed && callback) {
try {
await callback(String(chunk));
} catch (error) {
recordError(error);
callbackFailed = true;
}
}
// Even if a log sink fails, drain the pipe so the child can finish.
}
} catch (error) {
recordError(error);
}
};
const output = Promise.all([
consume(child.stdout, callbacks.onStdout),
consume(child.stderr, callbacks.onStderr),
]);
const completed = Promise.all([closed, ready, output]).then(([result]) => ({
...result,
error: failure,
}));
return { started, completed };
}
+11
View File
@@ -0,0 +1,11 @@
// Match command text literally: looking up a task must never execute its input.
export function findCronId(crontab: string, command: string): string {
if (!command) return '';
for (const line of crontab.split('\n')) {
const id = /\bID=(\d+)\s/.exec(line);
if (id && line.includes(command, id.index + id[0].length)) {
return id[1];
}
}
return '';
}
+110
View File
@@ -0,0 +1,110 @@
import CronExpressionParser from 'cron-parser';
import { validate } from 'node-cron';
import { ScheduleType } from '../interface/schedule';
const aliases: Record<string, string> = {
'@yearly': '0 0 0 1 1 *',
'@annually': '0 0 0 1 1 *',
'@monthly': '0 0 0 1 * *',
'@weekly': '0 0 0 * * 0',
'@daily': '0 0 0 * * *',
'@midnight': '0 0 0 * * *',
'@hourly': '0 0 * * * *',
'@minutely': '0 * * * * *',
};
export interface CronSchedule {
patterns: string[];
dayCount: number;
nthDay: number;
}
// Validation and registration repeatedly see identical rules in large snapshots.
// Cache only immutable syntax, never dates or running tasks, and bound user input.
const scheduleCache = new Map<string, CronSchedule>();
// Canonicalize legacy shorthand, aliases, names and numeric-start steps before
// node-cron sees them. Extend legacy aliases explicitly; reject H and bare /N.
export function parseCronSchedule(schedule: unknown): CronSchedule {
if (typeof schedule !== 'string' || !schedule.trim()) {
throw new Error('Invalid cron schedule');
}
let source = schedule.trim();
const cacheKey = source;
const cached = scheduleCache.get(cacheKey);
if (cached) return cached;
if (source.startsWith('@')) {
if (!Object.hasOwn(aliases, source)) throw new Error('Invalid cron alias');
source = aliases[source];
}
let parts = source.split(/\s+/);
if (
parts.length > 6 ||
parts.some((part) => /(^|,)\//.test(part) || /H(?:\(|\/|$)/i.test(part))
) {
throw new Error('Unsupported cron syntax');
}
parts = [
...['0', '*', '*', '*', '*', '*'].slice(0, 6 - parts.length),
...parts,
];
if (
parts.some(
(part, index) => index !== 3 && index !== 5 && part.includes('?'),
)
) {
throw new Error('Question mark is only valid in day fields');
}
const expression = CronExpressionParser.parse(parts.join(' '));
if (!expression.hasNext())
throw new Error('Cron schedule has no next execution');
const normalized = expression.stringify(true).replace(/\?/g, '*').split(' ');
const dayCount = expression.fields.dayOfMonth.values.length;
const weekCount = expression.fields.dayOfWeek.values.length;
const patterns =
dayCount < 31 && weekCount < 8
? [
[...normalized.slice(0, 5), '*'].join(' '),
[...normalized.slice(0, 3), '*', ...normalized.slice(4)].join(' '),
]
: [normalized.join(' ')];
if (!patterns.every((pattern) => validate(pattern))) {
throw new Error('Unsupported cron schedule');
}
const parsed = {
patterns,
dayCount,
nthDay: expression.fields.dayOfWeek.nthDay,
};
Object.freeze(patterns);
Object.freeze(parsed);
if (scheduleCache.size >= 512)
scheduleCache.delete(scheduleCache.keys().next().value);
scheduleCache.set(cacheKey, parsed);
return parsed;
}
export function isValidCronSchedule(schedule: unknown): schedule is string {
try {
parseCronSchedule(schedule);
return true;
} catch {
return false;
}
}
export function getInvalidCronSchedules(cron: {
schedule?: string;
extra_schedules?: Array<{ schedule: string }>;
}): unknown[] {
if (
cron.schedule?.startsWith(ScheduleType.ONCE) ||
cron.schedule?.startsWith(ScheduleType.BOOT)
) {
return [];
}
return [
cron.schedule,
...(cron.extra_schedules || []).map((x) => x.schedule),
].filter((schedule) => !isValidCronSchedule(schedule));
}
+89
View File
@@ -0,0 +1,89 @@
import { createTask, ScheduledTask } from 'node-cron';
import { parseCronSchedule } from './cronSchedule';
interface SchedulerLogger {
warn(message: string, ...args: unknown[]): unknown;
error(message: string, ...args: unknown[]): unknown;
}
export interface CronJob {
start(): void;
cancel(): void;
}
// node-cron owns timers and calendar calculation; Qinglong owns execution
// concurrency and process lifetime. Missed in-process slots join that same queue.
export function createCronJob(
schedule: string,
callback: (date: Date) => unknown | Promise<unknown>,
options: { name: string; logger: SchedulerLogger; start?: boolean },
): CronJob {
const parsed = parseCronSchedule(schedule);
const tasks: ScheduledTask[] = [];
let cancelled = false;
let started = false;
const accepts = (index: number, date: Date) => {
if (tasks.length === 1) return true;
// Preserve cron-parser 4's day/weekday OR semantics, including its
// month-dependent full-day range and the global nth-week constraint.
if (parsed.nthDay && Math.ceil(date.getDate() / 7) !== parsed.nthDay)
return false;
if (index === 1) return !tasks[0].match(date);
const monthDays = [31, 29, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31];
return parsed.dayCount < monthDays[date.getMonth()] || tasks[1].match(date);
};
const dispatch = async (index: number, date: Date, missed: boolean) => {
if (cancelled || !started || !accepts(index, date)) return;
if (missed) {
options.logger.warn(
'[schedule][补执行迟到任务] 任务: %s, 计划时间: %s',
options.name,
date.toISOString(),
);
}
try {
await callback(date);
} catch (error) {
options.logger.error(
'[schedule][定时回调失败] 任务: %s, 错误: %s',
options.name,
error instanceof Error ? error.message : String(error),
);
}
};
try {
for (const [index, pattern] of parsed.patterns.entries()) {
const task = createTask(
pattern,
(context) => dispatch(index, context.date, false),
{
noOverlap: false,
missedExecutionTolerance: 5000,
},
);
tasks.push(task);
task.on('execution:missed', (context) =>
dispatch(index, context.date, true),
);
// Fail before an existing schedule is cancelled, including impossible dates.
task.getNextRuns(1);
}
} catch (error) {
tasks.forEach((task) => task.destroy());
throw error;
}
const job: CronJob = {
start() {
if (started || cancelled) return;
started = true;
tasks.forEach((task) => task.start());
},
cancel() {
if (cancelled) return;
cancelled = true;
tasks.forEach((task) => task.destroy());
},
};
if (options.start !== false) job.start();
return job;
}
+4
View File
@@ -0,0 +1,4 @@
// Keep error formatting independent of database, HTTP and gRPC dependencies.
export function errStack(error: unknown): string {
return error instanceof Error && error.stack ? error.stack : String(error);
}
+7 -6
View File
@@ -3,6 +3,7 @@ import { shareStore } from './store';
const messages: Record<string, Record<string, string>> = {
zh: {},
en: {
'资源已存在,请检查重复的名称或值': 'Resource already exists; check for duplicate names or values',
'暂无权限': 'Access denied',
'参数错误': 'Invalid parameter',
'参数不正确': 'Invalid parameter',
@@ -94,8 +95,8 @@ const messages: Record<string, Record<string, string>> = {
'日志设置为忽略': 'Log set to ignore',
'定时规则不能为空': 'Schedule rule cannot be empty',
'无效的定时规则': 'Invalid schedule rule',
'日志名称只能包含字母、数字、下划线和连字符':
'Log name can only contain letters, numbers, underscores, and hyphens',
'日志名称只能包含中文、字母、数字、下划线、连字符、点和路径分隔符':
'Log name can only contain Chinese characters, letters, numbers, underscores, hyphens, dots, and path separators',
'日志名称不能超过100个字符': 'Log name cannot exceed 100 characters',
'错误的用户名密码,请重试': 'Incorrect username or password, please try again',
'该 IP 已被列入黑名单': 'This IP address has been blocked',
@@ -142,10 +143,10 @@ const messages: Record<string, Record<string, string>> = {
'Scheduler registration failed, task update rolled back',
'调度器注册失败,任务启用已回滚':
'Scheduler registration failed, task enable rolled back',
'任务ID %s: 无效的 cron 表达式 "%s"(不支持裸 /N 步长和 ? 字符)':
'Task ID %s: invalid cron expression "%s" (bare /N steps and ? character not supported)',
'任务ID %s (extra_schedule): 无效的 cron 表达式 "%s"(不支持裸 /N 步长和 ? 字符)':
'Task ID %s (extra_schedule): invalid cron expression "%s" (bare /N steps and ? character not supported)',
'任务ID %s: 无效的 cron 表达式 "%s"':
'Task ID %s: invalid cron expression "%s"',
'任务ID %s (extra_schedule): 无效的 cron 表达式 "%s"':
'Task ID %s (extra_schedule): invalid cron expression "%s"',
},
};
+1
View File
@@ -18,6 +18,7 @@ export interface IDependencyFn<T> {
export interface ICronFn<T> {
(): Promise<T>;
cron?: TCron;
isCurrent?: () => boolean;
}
export interface ISchedule {
+11
View File
@@ -0,0 +1,11 @@
import { resolveFileAccess } from './fileAccess';
/** Check stored paths too: old database rows are not trusted input. */
export function resolveLogPath(root: string, value: string): string {
const target = resolveFileAccess(root, [value]);
if (!target)
throw Object.assign(new Error('Log path is outside the log directory'), {
status: 400,
});
return target;
}
+87 -73
View File
@@ -1,5 +1,8 @@
import { createWriteStream, WriteStream } from 'fs';
import { EventEmitter } from 'events';
import path from 'path';
import config from '../config';
import { resolveFileAccess } from './fileAccess';
/**
* Manages write streams for log files to improve performance by avoiding repeated file opens
@@ -8,83 +11,90 @@ export class LogStreamManager extends EventEmitter {
private streams: Map<string, WriteStream> = new Map();
private pendingWrites: Map<string, Promise<void>> = new Map();
/**
* Write data to a log file using a managed stream
* @param filePath - Absolute path to the log file
* @param data - Data to write to the log file
*/
async write(filePath: string, data: string): Promise<void> {
// Wait for any pending writes to this file to complete
const pending = this.pendingWrites.get(filePath);
if (pending) {
await pending;
}
private closingStreams = new Map<string, Promise<void>>();
private closedStreams = new WeakSet<WriteStream>();
private streamErrors = new Map<string, Error>();
// Create a new promise for this write operation
const writePromise = new Promise<void>((resolve, reject) => {
let stream = this.streams.get(filePath);
if (!stream) {
// Create a new write stream if one doesn't exist
stream = createWriteStream(filePath, { flags: 'a' });
this.streams.set(filePath, stream);
// Handle stream errors
stream.on('error', (error) => {
this.emit('error', { filePath, error });
// Remove the stream from the map on error
this.streams.delete(filePath);
reject(error);
});
}
// Write the data
const canContinue = stream.write(data, 'utf8', (error) => {
if (error) {
reject(error);
} else {
resolve();
}
});
// Handle backpressure
if (!canContinue) {
stream.once('drain', () => {
// Stream is ready for more data
});
}
});
this.pendingWrites.set(filePath, writePromise);
try {
await writePromise;
} finally {
this.pendingWrites.delete(filePath);
}
constructor(private readonly logRoot = config.logPath) {
super();
}
/**
* Close the stream for a specific file path
* @param filePath - Absolute path to the log file
*/
async closeStream(filePath: string): Promise<void> {
// Wait for any pending writes to complete
const pending = this.pendingWrites.get(filePath);
if (pending) {
await pending.catch(() => {
// Ignore errors on pending writes during close
});
/** Register each write synchronously, so concurrent callers cannot lose the tail. */
async write(filePath: string, data: string): Promise<void> {
if (this.closingStreams.has(filePath)) {
throw new Error(`Log stream is closing: ${filePath}`);
}
const previous = this.pendingWrites.get(filePath) || Promise.resolve();
const pending = previous.then(
() =>
new Promise<void>((resolve, reject) => {
const failure = this.streamErrors.get(filePath);
if (failure) return reject(failure);
let stream = this.streams.get(filePath);
if (!stream) {
// Validate only when opening: subsequent chunks reuse the same descriptor.
const root = path.resolve(this.logRoot);
const target = path.resolve(filePath);
if (
!target.startsWith(root + path.sep) ||
!resolveFileAccess(root, [target])
) {
return reject(new Error('Log path is outside the log directory'));
}
stream = createWriteStream(target, { flags: 'a' });
this.streams.set(filePath, stream);
const current = stream;
stream.once('close', () => this.closedStreams.add(current));
stream.on('error', (error) => {
this.streamErrors.set(filePath, error);
// EventEmitter's unobserved "error" event would crash the caller.
if (this.listenerCount('error') > 0)
this.emit('error', { filePath, error });
});
}
stream.write(data, 'utf8', (error) =>
error ? reject(error) : resolve(),
);
}),
);
this.pendingWrites.set(filePath, pending);
// Keep the tail until close, including failures; never reopen a failed log mid-run.
return pending;
}
const stream = this.streams.get(filePath);
if (stream) {
return new Promise<void>((resolve) => {
stream.end(() => {
this.streams.delete(filePath);
resolve();
});
});
async closeStream(filePath: string): Promise<void> {
const closing = this.closingStreams.get(filePath);
if (closing) return closing;
const pending = this.pendingWrites.get(filePath);
const result = (async () => {
let failure: unknown;
try {
await pending;
} catch (error) {
failure = error;
}
const stream = this.streams.get(filePath);
try {
if (stream && !this.closedStreams.has(stream)) {
await new Promise<void>((resolve) => {
stream.once('close', resolve);
if (failure || stream.destroyed) stream.destroy();
else stream.end();
});
}
failure ||= this.streamErrors.get(filePath);
if (failure) throw failure;
} finally {
this.streams.delete(filePath);
this.pendingWrites.delete(filePath);
this.streamErrors.delete(filePath);
}
})();
this.closingStreams.set(filePath, result);
try {
await result;
} finally {
this.closingStreams.delete(filePath);
}
}
@@ -92,7 +102,11 @@ export class LogStreamManager extends EventEmitter {
* Close all open streams
*/
async closeAll(): Promise<void> {
const closePromises = Array.from(this.streams.keys()).map((filePath) =>
const paths = new Set([
...this.streams.keys(),
...this.pendingWrites.keys(),
]);
const closePromises = Array.from(paths).map((filePath) =>
this.closeStream(filePath),
);
await Promise.all(closePromises);
+47
View File
@@ -0,0 +1,47 @@
const RETRY_TTL = 15 * 60 * 1000;
const MAX_SOURCES = 10000;
const MAX_RETRY_SECONDS = 5 * 60;
/** Source-specific, bounded backoff; persisted account history is not a lock. */
export class LoginThrottle {
private sources = new Map<
string,
{ failures: number; lastFailure: number }
>();
private get(source: string, now: number) {
const state = this.sources.get(source);
if (state && now - state.lastFailure >= RETRY_TTL) {
this.sources.delete(source);
return undefined;
}
return state;
}
retryAfter(source: string, now: number): number {
const state = this.get(source, now);
if (!state || state.failures < 3) return 0;
const seconds = Math.min(3 ** state.failures, MAX_RETRY_SECONDS);
return Math.max(
0,
Math.ceil((seconds * 1000 - (now - state.lastFailure)) / 1000),
);
}
fail(source: string, now: number): number {
const failures = Math.min((this.get(source, now)?.failures || 0) + 1, 6);
this.sources.delete(source);
if (this.sources.size >= MAX_SOURCES) {
this.sources.delete(this.sources.keys().next().value!);
}
this.sources.set(source, { failures, lastFailure: now });
return failures;
}
reset(source: string) {
this.sources.delete(source);
}
clear() {
this.sources.clear();
}
}
+16 -9
View File
@@ -50,11 +50,9 @@ class TaskLimit {
Buffer.from(tlsConfig.clientCert),
)
: credentials.createInsecure();
this._client = new ApiClient(
`localhost:${config.grpcPort}`,
creds,
{ 'grpc.enable_http_proxy': 0 },
);
this._client = new ApiClient(`localhost:${config.grpcPort}`, creds, {
'grpc.enable_http_proxy': 0,
});
}
return this._client;
}
@@ -113,12 +111,15 @@ class TaskLimit {
}
}
public removeQueuedCron(id: string) {
public removeQueuedCron(id: string, completed?: ICronFn<any>) {
if (this.queuedCrons.has(id)) {
const runs = this.queuedCrons.get(id);
if (runs && runs.length > 0) {
runs.pop();
this.queuedCrons.set(id, runs);
const remaining = completed
? runs.filter((run) => run !== completed)
: runs.slice(0, -1);
if (remaining.length) this.queuedCrons.set(id, remaining);
else this.queuedCrons.delete(id);
}
}
}
@@ -143,9 +144,15 @@ class TaskLimit {
cron: TCron,
fn: ICronFn<T>,
options?: Partial<QueueAddOptions>,
isCurrent: () => boolean = () => true,
): Promise<T | void> {
if (!isCurrent()) return;
fn.cron = cron;
let runs = this.queuedCrons.get(cron.id);
fn.isCurrent = isCurrent;
// Invalidated snapshots must not consume the new revision's repeat budget.
const runs = this.queuedCrons
.get(cron.id)
?.filter((run) => run.isCurrent?.() !== false);
const result = runs?.length ? [...runs, fn] : [fn];
const repeatTimes = this.repeatCronNotifyMap.get(cron.id) || 0;
if (result?.length > 5) {
+53 -34
View File
@@ -4,21 +4,27 @@ import Logger from '../loaders/logger';
import { ICron } from '../protos/cron';
import { CrontabModel, CrontabStatus } from '../data/cron';
import { killTask } from '../config/util';
import {
RunningInstanceModel,
InstanceStatus,
} from '../data/runningInstance';
import { RunningInstanceModel, InstanceStatus } from '../data/runningInstance';
import dayjs from 'dayjs';
import { observeChildProcess, asError } from './childProcess';
export function runCron(cmd: string, cron: ICron): Promise<number | void> {
return taskLimit.runWithCronLimit(cron, () => {
return new Promise(async (resolve: any) => {
export function runCron(
cmd: string,
cron: ICron,
isCurrent: () => boolean = () => true,
): Promise<number | void> {
const execute = async () => {
try {
if (!isCurrent()) return;
// Check if the cron is already running and stop it (only if multiple instances are not allowed)
try {
const existingCron = await CrontabModel.findOne({
where: { id: Number(cron.id) },
});
// A mutation may commit while this queued run is reading the database.
if (!isCurrent() || !existingCron || existingCron.isDisabled) return;
// Default to single instance mode (0) for backward compatibility
const allowSingleInstances =
existingCron?.allow_multiple_instances === 0;
@@ -33,12 +39,17 @@ export function runCron(cmd: string, cron: ICron): Promise<number | void> {
Logger.info(
`[schedule][停止已运行任务] 任务ID: ${cron.id}, PID: ${existingCron.pid}`,
);
await killTask(existingCron.pid);
await killTask(existingCron.pid, true);
// Mark old running instances as stopped
const stoppedAt = dayjs().unix();
await RunningInstanceModel.update(
{ status: InstanceStatus.stopped, finished_at: stoppedAt },
{ where: { cron_id: Number(cron.id), status: InstanceStatus.running } },
{
where: {
cron_id: Number(cron.id),
status: InstanceStatus.running,
},
},
);
// Update the status to idle after killing
await CrontabModel.update(
@@ -50,8 +61,11 @@ export function runCron(cmd: string, cron: ICron): Promise<number | void> {
Logger.error(
`[schedule][检查已运行任务失败] 任务ID: ${cron.id}, 错误: ${error}`,
);
throw error;
}
// In particular, invalidation can happen while stopping a previous run.
if (!isCurrent()) return;
Logger.info(
`[schedule][开始执行任务] 参数 ${JSON.stringify({
...cron,
@@ -60,33 +74,38 @@ export function runCron(cmd: string, cron: ICron): Promise<number | void> {
);
const cp = spawn(cmd, { shell: '/bin/bash' });
cp.stderr.on('data', (data) => {
Logger.info(
'[schedule][执行任务失败] 命令: %s, 错误信息: %j',
cmd,
data.toString(),
);
const { completed } = observeChildProcess(cp, {
onStderr: async (message) => {
Logger.info(
'[schedule][任务标准错误] 命令: %s, 信息: %s',
cmd,
message,
);
},
});
cp.on('error', (err) => {
const result = await completed;
if (result.error) {
Logger.error(
'[schedule][创建任务失败] 命令: %s, 错误信息: %j',
'[schedule][执行任务失败] 命令: %s, 错误: %s',
cmd,
err,
result.error.message,
);
});
cp.on('exit', async (code) => {
taskLimit.removeQueuedCron(cron.id);
Logger.info(
'[schedule][执行任务结束] 参数: %s, 退出码: %j',
JSON.stringify({
...cron,
command: cmd,
}),
code,
);
resolve({ ...cron, command: cmd, pid: cp.pid, code });
});
});
});
}
Logger.info(
'[schedule][执行任务结束] 任务ID: %s, 退出码: %j',
cron.id,
result.code,
);
return { ...cron, command: cmd, pid: cp.pid, ...result } as any;
} catch (error) {
Logger.error(
'[schedule][创建任务失败] 命令: %s, 错误: %s',
cmd,
asError(error).message,
);
} finally {
taskLimit.removeQueuedCron(cron.id, execute);
}
};
return taskLimit.runWithCronLimit(cron, execute, undefined, isCurrent);
}
+40
View File
@@ -0,0 +1,40 @@
import lockfile from 'proper-lockfile';
import config from '../config';
// HTTP and gRPC both mutate cron definitions. Hold one shared lock from the
// initial DB read/write through scheduler registration (including rollback).
// Recovery takes the same lock before reading its replacement snapshot.
export async function withSchedulerMutation<T>(
operation: () => Promise<T>,
): Promise<T> {
let release: () => Promise<void>;
try {
// proper-lockfile indexes held locks by target, not lockfilePath. Keep
// this identity separate from nested writeFileWithLock(crontabFile).
release = await lockfile.lock(`${config.crontabFile}.scheduler`, {
realpath: false,
lockfilePath: `${config.crontabFile}.scheduler.lock`,
stale: 30000,
update: 10000,
retries: { retries: 50, factor: 1, minTimeout: 100, maxTimeout: 100 },
});
} catch (cause) {
throw Object.assign(
new Error('Scheduler configuration is busy', { cause }),
{
status: 503,
},
);
}
try {
return await operation();
} finally {
await release();
}
}
export function schedulerRegistrationError(message: string, cause: any): Error {
return Object.assign(new Error(message, { cause }), {
status: cause?.status === 503 ? 503 : 500,
});
}
+77
View File
@@ -0,0 +1,77 @@
// Recovery is single-flight and retried only while unavailable, never at idle.
export class SchedulerReadiness {
private ready = false;
private generation = 0;
private restore?: () => Promise<void>;
private pending?: Promise<boolean>;
private retry?: NodeJS.Timeout;
constructor(private probe: () => Promise<void>, private retryMs = 1000) {}
configure(restore: () => Promise<void>) {
this.restore = restore;
}
invalidate() {
this.ready = false;
this.generation++;
void this.recover();
}
recover(): Promise<boolean> {
if (this.pending) return this.pending;
if (!this.restore) return Promise.resolve(false);
clearTimeout(this.retry);
const generation = this.generation;
this.ready = false;
this.pending = (async () => {
try {
await this.probe();
await this.restore!();
await this.probe();
this.ready = generation === this.generation;
} catch {
this.ready = false;
}
return this.ready;
})().finally(() => {
this.pending = undefined;
if (!this.ready) {
this.retry = setTimeout(() => void this.recover(), this.retryMs);
this.retry.unref();
}
});
return this.pending;
}
async check(): Promise<boolean> {
if (!this.ready) return false;
const generation = this.generation;
try {
await this.probe();
return this.ready && generation === this.generation;
} catch {
this.invalidate();
return false;
}
}
async ensureReady(timeoutMs = 2000): Promise<void> {
let timer: NodeJS.Timeout | undefined;
try {
const available = await Promise.race([
(async () => (await this.check()) || (await this.recover()))(),
new Promise<boolean>((resolve) => {
timer = setTimeout(() => resolve(false), timeoutMs);
}),
]);
if (!available) {
throw Object.assign(new Error('Scheduler is recovering; try again later'), {
status: 503,
});
}
} finally {
clearTimeout(timer);
}
}
}
+70
View File
@@ -0,0 +1,70 @@
import { QueryTypes, Sequelize } from 'sequelize';
// Append new entries; IDs are persisted and must not be renumbered or reused.
const columns = [
{
table: 'CrontabViews',
column: 'filterRelation',
type: 'VARCHAR(255)',
},
{ table: 'Subscriptions', column: 'proxy', type: 'VARCHAR(255)' },
{ table: 'CrontabViews', column: 'type', type: 'NUMBER' },
{ table: 'Subscriptions', column: 'autoAddCron', type: 'NUMBER' },
{ table: 'Subscriptions', column: 'autoDelCron', type: 'NUMBER' },
{ table: 'Crontabs', column: 'sub_id', type: 'NUMBER' },
{ table: 'Crontabs', column: 'extra_schedules', type: 'JSON' },
{ table: 'Crontabs', column: 'task_before', type: 'TEXT' },
{ table: 'Crontabs', column: 'task_after', type: 'TEXT' },
{ table: 'Crontabs', column: 'log_name', type: 'VARCHAR(255)' },
{
table: 'Crontabs',
column: 'allow_multiple_instances',
type: 'NUMBER',
},
{ table: 'Crontabs', column: 'work_dir', type: 'VARCHAR(255)' },
{ table: 'Envs', column: 'isPinned', type: 'NUMBER' },
{ table: 'Envs', column: 'labels', type: 'JSON' },
{ table: 'Crontabs', column: 'queued_token', type: 'VARCHAR(255)' },
];
export async function migrateSchema(database: Sequelize): Promise<void> {
await database.transaction(async (transaction) => {
await database.query(
'CREATE TABLE IF NOT EXISTS "SchemaMigrations" ("id" TEXT PRIMARY KEY, "applied_at" TEXT NOT NULL)',
{ transaction },
);
const applied = await database.query<{ id: string }>(
'SELECT "id" FROM "SchemaMigrations"',
{ type: QueryTypes.SELECT, transaction },
);
const appliedIds = new Set(applied.map(({ id }) => id));
for (const { table, column, type } of columns) {
const id = `add-${table}-${column}`;
const fields = await database.query<{ name: string }>(
`PRAGMA table_info("${table}")`,
{
type: QueryTypes.SELECT,
transaction,
},
);
if (fields.length === 0)
throw new Error(`Migration table is missing: ${table}`);
if (!fields.some((field) => field.name === column)) {
// table/column/type come only from the static migration manifest above.
await database.query(
`ALTER TABLE "${table}" ADD COLUMN "${column}" ${type}`,
{ transaction },
);
}
if (!appliedIds.has(id)) {
await database.query(
'INSERT INTO "SchemaMigrations" ("id", "applied_at") VALUES (:id, :appliedAt)',
{
replacements: { id, appliedAt: new Date().toISOString() },
transaction,
},
);
}
}
});
}
+467
View File
@@ -0,0 +1,467 @@
import * as fs from 'fs/promises';
import { createReadStream } from 'fs';
import path from 'path';
import { createHash, randomUUID } from 'crypto';
import { promisify } from 'util';
import { brotliCompress, brotliDecompress, constants } from 'zlib';
import { lock } from 'proper-lockfile';
import { resolveFileAccess } from './fileAccess';
import { getUniqueLockPath } from './utils';
const compress = promisify(brotliCompress);
const decompress = promisify(brotliDecompress);
const hash = (content: string) =>
createHash('sha256').update(content).digest('hex');
export const SCRIPT_HISTORY_LIMITS = {
versions: 20,
fileBytes: 1024 * 1024,
unpackedBytes: 8 * 1024 * 1024,
archiveBytes: 2 * 1024 * 1024,
totalBytes: 32 * 1024 * 1024,
files: 1000,
};
type Source = 'baseline' | 'save' | 'restore' | 'external';
interface Version {
id: string;
createdAt: string;
source: Source;
hash: string;
content: string;
}
interface Archive {
format: 1;
path: string;
versions: Version[];
head?: { hash: string; createdAt: string; source: Source };
}
export class ScriptHistoryError extends Error {
constructor(public status: number, message: string) {
super(message);
}
}
export class HistoryUnavailableError extends ScriptHistoryError {
constructor(public currentHash: string) {
super(413, '历史版本仅支持不超过 1 MiB 的 UTF-8 文本脚本');
}
}
interface SaveOptions {
skipHistory?: boolean;
expectedHash?: string;
}
/** Lazy, bounded storage: no watcher, database, daemon, or in-memory cache.
* Full snapshots share a Brotli window, so similar revisions compress together
* without a patch chain. All mutations use the existing script lock and a
* cross-process archive lock. History is durable before overwriting the script. */
export class ScriptHistory {
constructor(
private scriptRoot: string,
private historyRoot: string,
private blacklist: string[] = [],
private limits = SCRIPT_HISTORY_LIMITS,
) {}
private async target(directory: string, filename: string) {
const candidate = resolveFileAccess(
this.scriptRoot,
[directory, filename],
this.blacklist,
);
if (!candidate) throw new ScriptHistoryError(403, '暂无权限');
try {
const file = await fs.realpath(candidate);
const root = await fs.realpath(this.scriptRoot);
const relative = path.relative(root, file);
const key = hash(relative);
return {
file,
relative,
archive: path.join(this.historyRoot, key + '.br'),
};
} catch (error: any) {
if (error.code === 'ENOENT')
throw new ScriptHistoryError(404, '脚本不存在');
throw error;
}
}
private async readText(file: string) {
const handle = await fs.open(file, 'r');
try {
const stat = await handle.stat();
if (!stat.isFile()) throw new ScriptHistoryError(400, '请选择脚本文件');
if (stat.size > this.limits.fileBytes)
throw new ScriptHistoryError(
413,
'历史版本仅支持不超过 1 MiB 的 UTF-8 文本脚本',
);
// Bounded read even if an external process grows the file after stat.
const bytes = Buffer.alloc(this.limits.fileBytes + 1);
let size = 0;
while (size < bytes.length) {
const { bytesRead } = await handle.read(
bytes,
size,
bytes.length - size,
null,
);
if (!bytesRead) break;
size += bytesRead;
}
const buffer = bytes.subarray(0, size);
const content = buffer.toString('utf8');
if (
size > this.limits.fileBytes ||
content.includes('\0') ||
!Buffer.from(content).equals(buffer)
) {
throw new ScriptHistoryError(
413,
'历史版本仅支持不超过 1 MiB 的 UTF-8 文本脚本',
);
}
return content;
} finally {
await handle.close();
}
}
private async readArchive(file: string, relative: string): Promise<Archive> {
try {
const stat = await fs.lstat(file);
if (!stat.isFile() || stat.size > this.limits.archiveBytes)
throw new Error('Invalid archive');
const bytes = await decompress(await fs.readFile(file), {
maxOutputLength: this.limits.unpackedBytes,
});
const archive: Archive = JSON.parse(bytes.toString('utf8'));
if (
archive.format !== 1 ||
archive.path !== relative ||
!Array.isArray(archive.versions) ||
archive.versions.length > this.limits.versions ||
archive.versions.some(
(v) =>
typeof v.content !== 'string' ||
typeof v.id !== 'string' ||
typeof v.createdAt !== 'string' ||
!['baseline', 'save', 'restore', 'external'].includes(v.source) ||
Buffer.byteLength(v.content) > this.limits.fileBytes ||
hash(v.content) !== v.hash,
)
) {
throw new Error('Invalid archive');
}
return archive;
} catch (error: any) {
if (error.code === 'ENOENT')
return { format: 1, path: relative, versions: [] };
throw new ScriptHistoryError(500, '历史版本读取失败,请检查历史文件');
}
}
async list(directory: string, filename: string) {
const target = await this.target(directory, filename);
const [current, archive] = await Promise.all([
this.readText(target.file),
this.readArchive(target.archive, target.relative),
]);
const currentHash = hash(current);
const versions = archive.versions
.slice()
.reverse()
.map(({ content, ...v }) => ({
...v,
size: Buffer.byteLength(content),
identical: v.hash === currentHash,
}));
return { versions, currentHash, limit: this.limits.versions };
}
async detail(directory: string, filename: string, id: string) {
const target = await this.target(directory, filename);
const [current, archive] = await Promise.all([
this.readText(target.file),
this.readArchive(target.archive, target.relative),
]);
const version = archive.versions.find((v) => v.id === id);
if (!version) throw new ScriptHistoryError(404, '历史版本不存在或已清理');
return { version, current, currentHash: hash(current) };
}
async save(
directory: string,
filename: string,
content: string,
options: SaveOptions = {},
) {
return this.mutate(directory, filename, { content, ...options });
}
async restore(
directory: string,
filename: string,
id: string,
expectedHash: string,
) {
return this.mutate(directory, filename, { id, expectedHash });
}
private async mutate(
directory: string,
filename: string,
change:
| ({ content: string } & SaveOptions)
| { id: string; expectedHash: string },
) {
const target = await this.target(directory, filename);
const releaseFile = await lock(target.file, {
lockfilePath: getUniqueLockPath(target.file),
retries: { retries: 20, minTimeout: 50, maxTimeout: 250 },
});
try {
let current: string;
try {
current = await this.readText(target.file);
if (
'content' in change &&
(Buffer.byteLength(change.content) > this.limits.fileBytes ||
change.content.includes('\0'))
) {
throw new ScriptHistoryError(
413,
'历史版本仅支持不超过 1 MiB 的 UTF-8 文本脚本',
);
}
} catch (error) {
if (
!('content' in change) ||
!(error instanceof ScriptHistoryError) ||
error.status !== 413
)
throw error;
// Hash oversized or non-UTF-8 files without reading them all into memory.
const digest = createHash('sha256');
for await (const chunk of createReadStream(target.file))
digest.update(chunk);
const currentHash = digest.digest('hex');
if (!change.skipHistory) throw new HistoryUnavailableError(currentHash);
if (change.expectedHash !== currentHash)
throw new ScriptHistoryError(409, '文件已变化,请刷新历史版本后重试');
await this.preparedWrite(target.file, change.content, (commit) =>
commit(),
);
return {
content: change.content,
changed: true,
historyRecorded: false,
};
}
if (
('id' in change || change.skipHistory) &&
hash(current) !== change.expectedHash
) {
throw new ScriptHistoryError(409, '文件已变化,请刷新历史版本后重试');
}
if ('content' in change && change.content === current)
return { content: current, changed: false };
await fs.mkdir(this.historyRoot, { recursive: true, mode: 0o700 });
const releaseHistory = await lock(this.historyRoot, {
retries: { retries: 20, minTimeout: 50, maxTimeout: 250 },
});
try {
const archive = await this.readArchive(target.archive, target.relative);
const restored =
'id' in change
? archive.versions.find((v) => v.id === change.id)
: undefined;
if ('id' in change && !restored)
throw new ScriptHistoryError(404, '历史版本不存在或已清理');
const content =
'content' in change ? change.content : restored!.content;
if (content === current) return { content, changed: false };
if (
Buffer.byteLength(content) > this.limits.fileBytes ||
content.includes('\0')
) {
throw new ScriptHistoryError(
413,
'历史版本仅支持不超过 1 MiB 的 UTF-8 文本脚本',
);
}
const currentHash = hash(current);
if (
archive.versions[archive.versions.length - 1]?.hash !== currentHash
) {
const head =
archive.head?.hash === currentHash ? archive.head : undefined;
archive.versions.push({
id: randomUUID(),
createdAt: head?.createdAt || new Date().toISOString(),
source:
head?.source ||
(archive.versions.length ? 'external' : 'baseline'),
hash: currentHash,
content: current,
});
}
// Only prior contents are stored. Head metadata labels the next snapshot,
// but is used only if its hash matches the file. A failed file write can
// never introduce a phantom content version or lose the old snapshot.
archive.head = {
hash: hash(content),
createdAt: new Date().toISOString(),
source: 'id' in change ? 'restore' : 'save',
};
const packed = await this.pack(archive);
const cleanupPending = await this.preparedWrite(
target.file,
content,
(commit) => this.store(target.archive, packed, commit),
);
return {
content,
changed: true,
historyRecorded: true,
cleanupPending,
};
} finally {
await releaseHistory();
}
} finally {
await releaseFile();
}
}
private async pack(archive: Archive) {
archive.versions = archive.versions.slice(-this.limits.versions);
while (true) {
const serialized = Buffer.from(JSON.stringify(archive));
if (serialized.length <= this.limits.unpackedBytes) {
const compressed = await compress(serialized, {
params: {
[constants.BROTLI_PARAM_QUALITY]: 4,
[constants.BROTLI_PARAM_LGWIN]: 22,
},
});
if (compressed.length <= this.limits.archiveBytes) return compressed;
}
if (archive.versions.length <= 1)
throw new ScriptHistoryError(413, '历史版本容量不足,无法安全保存');
archive.versions.shift();
}
}
/** Stage the script before publishing history. A failed write/fsync never
* truncates the original. Preserve the existing file's ownership and mode. */
private async preparedWrite<T>(
file: string,
content: string,
run: (commit: () => Promise<void>) => Promise<T>,
): Promise<T> {
const stat = await fs.stat(file);
const temporary = path.join(
path.dirname(file),
`.ql-save-${randomUUID()}.tmp`,
);
try {
await fs.writeFile(temporary, content, {
encoding: 'utf8',
mode: 0o600,
flag: 'wx',
});
if (
typeof process.getuid === 'function' &&
typeof process.getgid === 'function' &&
(stat.uid !== process.getuid() || stat.gid !== process.getgid())
) {
await fs.chown(temporary, stat.uid, stat.gid);
}
await fs.chmod(temporary, stat.mode & 0o7777);
const handle = await fs.open(temporary, 'r');
try {
await handle.sync();
} finally {
await handle.close();
}
return await run(() => fs.rename(temporary, file));
} finally {
await fs.unlink(temporary).catch(() => undefined);
}
}
private async store(
file: string,
content: Buffer,
commit: () => Promise<void>,
) {
const entries = [];
for (const name of await fs.readdir(this.historyRoot)) {
if (!/^[a-f0-9]{64}\.br$/.test(name)) continue;
const fullPath = path.join(this.historyRoot, name);
if (fullPath === file) continue;
const stat = await fs.lstat(fullPath);
entries.push({ file: fullPath, size: stat.size, time: stat.mtimeMs });
}
entries.sort((a, b) => a.time - b.time);
let total = entries.reduce((sum, v) => sum + v.size, content.length);
const victims: string[] = [];
while (
entries.length &&
(total > this.limits.totalBytes || entries.length >= this.limits.files)
) {
const oldest = entries.shift()!;
victims.push(oldest.file);
total -= oldest.size;
}
if (total > this.limits.totalBytes)
throw new ScriptHistoryError(413, '历史版本容量不足,无法安全保存');
const previous = await fs.readFile(file).catch((error) => {
if (error.code === 'ENOENT') return undefined;
throw error;
});
const temporary = path.join(this.historyRoot, '.pending.tmp');
const publish = async (bytes: Buffer) => {
await fs.unlink(temporary).catch((error) => {
if (error.code !== 'ENOENT') throw error;
});
await fs.writeFile(temporary, bytes, { mode: 0o600, flag: 'wx' });
const handle = await fs.open(temporary, 'r');
try {
await handle.sync();
} finally {
await handle.close();
}
await fs.rename(temporary, file);
};
try {
await publish(content);
try {
await commit();
} catch (error) {
// The original script is still intact. Restore its exact history before
// returning an error; no unrelated archive has been removed yet.
if (previous) await publish(previous);
else await fs.unlink(file);
throw error;
}
// Quota eviction is post-commit maintenance. Report cleanup failures as a
// warning, not a failed save (which would encourage a misleading retry).
let cleanupPending = false;
for (const victim of victims) {
try {
await fs.unlink(victim);
} catch (error: any) {
if (error.code !== 'ENOENT') cleanupPending = true;
}
}
return cleanupPending;
} finally {
await fs.unlink(temporary).catch(() => undefined);
}
}
}
+5
View File
@@ -0,0 +1,5 @@
/** Quote one literal POSIX shell argument, including quotes and substitutions. */
export function shellQuote(value: string): string {
if (value.includes('\0')) throw new Error('Invalid shell argument');
return "'" + value.replace(/'/g, "'\\''") + "'";
}
+47
View File
@@ -0,0 +1,47 @@
import { fork } from 'child_process';
// Keep one-off startup dependencies out of the long-lived cluster primary.
export function runStartupProcess(entrypoint: string): Promise<void> {
return new Promise((resolve, reject) => {
const child = fork(entrypoint, [], { stdio: 'inherit' });
let interrupted: NodeJS.Signals | undefined;
let killTimer: NodeJS.Timeout | undefined;
const terminate = (signal: NodeJS.Signals) => {
if (interrupted) return;
interrupted = signal;
child.kill(signal);
killTimer = setTimeout(() => child.kill('SIGKILL'), 8000);
killTimer.unref();
};
const onSigterm = () => terminate('SIGTERM');
const onSigint = () => terminate('SIGINT');
const onExit = () => child.kill('SIGKILL');
const cleanup = () => {
if (killTimer) clearTimeout(killTimer);
process.removeListener('SIGTERM', onSigterm);
process.removeListener('SIGINT', onSigint);
process.removeListener('exit', onExit);
};
process.once('SIGTERM', onSigterm);
process.once('SIGINT', onSigint);
process.once('exit', onExit);
child.once('error', (error) => {
cleanup();
reject(error);
});
child.once('close', (code, signal) => {
cleanup();
if (code === 0 && !interrupted) {
resolve();
} else {
reject(
new Error(
`Startup process failed (${interrupted || signal || code})`,
),
);
}
});
});
}
+30
View File
@@ -0,0 +1,30 @@
import lockfile from 'proper-lockfile';
import path from 'path';
import config from '../config';
// Protect subscription storage ownership from the first database read through
// cleanup. Creation and URL/branch updates must use the same cross-process lock.
export async function withSubscriptionMutation<T>(
operation: () => Promise<T>,
): Promise<T> {
const target = `${path.resolve(config.scriptPath)}.subscriptions`;
let release: () => Promise<void>;
try {
release = await lockfile.lock(target, {
realpath: false,
stale: 30000,
update: 10000,
retries: { retries: 50, factor: 1, minTimeout: 100, maxTimeout: 100 },
});
} catch (cause) {
throw Object.assign(
new Error('Subscription configuration is busy', { cause }),
{ status: 503 },
);
}
try {
return await operation();
} finally {
await release();
}
}
+45
View File
@@ -0,0 +1,45 @@
import { resolveFileAccess } from './fileAccess';
import { createHash } from 'crypto';
export function assertSubscriptionAlias(alias: string): void {
if (
typeof alias !== 'string' ||
alias.length > 200 ||
!/^[^\s\\\p{C}]+$/u.test(alias) ||
alias.startsWith('~') ||
alias.split('/').some((part) => !part || part === '.' || part === '..')
) {
throw Object.assign(new Error('Invalid subscription alias'), {
status: 400,
});
}
}
/** Keep ordinary SSH names and use a reserved namespace for path-like aliases.
* The database alias and its script/log directories remain unchanged. */
export function getSubscriptionSshAlias(alias: string): string {
assertSubscriptionAlias(alias);
if (/^[\p{L}\p{N}_.+-]+$/u.test(alias) && !alias.endsWith('.config')) {
return alias;
}
return `~sub_${createHash('sha256').update(alias).digest('hex')}`;
}
export function resolveSubscriptionPath(
root: string,
alias: string,
suffix = '',
): string {
// Internal SSH filenames cannot collide with user-supplied aliases.
if (!/^~sub_[a-f0-9]{64}$/.test(alias)) {
assertSubscriptionAlias(
alias.startsWith('~global_') ? alias.slice(8) : alias,
);
}
const target = resolveFileAccess(root, [alias + suffix]);
if (!target)
throw Object.assign(new Error('Invalid subscription path'), {
status: 400,
});
return target;
}
+106
View File
@@ -0,0 +1,106 @@
import type { Subscription } from '../data/subscription';
import { formatUrl } from '../config/subscription';
import path from 'path';
import fs from 'fs';
import { resolveFileAccess } from './fileAccess';
// Storage namespaces come from URLs and branches, not user aliases. In
// particular, a legal Git directory can exceed the alias's 200-character limit.
export function isSubscriptionStorageName(name: string): boolean {
return (
typeof name === 'string' &&
!!name &&
!path.isAbsolute(name) &&
!/[\0\r\n\\]/.test(name) &&
!name.split('/').some((part) => !part || part === '.' || part === '..')
);
}
export function resolveSubscriptionStoragePath(
root: string,
name: string,
): string | undefined {
const target = isSubscriptionStorageName(name)
? resolveFileAccess(root, [name])
: '';
if (!target) {
if (isSubscriptionStorageName(name)) {
try {
fs.lstatSync(path.resolve(root, name));
} catch (error: any) {
// A long query can make the Shell candidate unrepresentable while
// the CLI's pathname-based file exists. Use the actual filesystem
// limit: macOS permits Unicode names exceeding 255 UTF-8 bytes.
if (error.code === 'ENAMETOOLONG') return undefined;
}
}
throw Object.assign(new Error('Invalid subscription path'), {
status: 400,
});
}
return target;
}
/** Match shell/update.sh get_uniq_path and update_raw, including SCP URLs. */
export function getSubscriptionStorageNames(
doc: Subscription,
engine: 'shell' | 'cli' = 'shell',
): {
script: string;
repo?: string;
raw?: string;
} {
// Old imported rows may only have an alias and no URL.
if (!doc.url) return { script: doc.alias, repo: doc.alias };
const url = formatUrl(doc).url!;
const trimmed = url.endsWith('/') ? url.slice(0, -1) : url;
const basename = trimmed.slice(trimmed.lastIndexOf('/') + 1);
const dot = basename.lastIndexOf('.');
const repo = dot < 0 ? basename : basename.slice(0, dot);
const parentUrl = engine === 'cli' ? trimmed : url;
const slash = parentUrl.lastIndexOf('/');
const parent =
engine === 'shell' && slash < 0 ? url : parentUrl.slice(0, slash);
const owner = parent.slice(parent.lastIndexOf('/') + 1);
const author = owner.slice(owner.lastIndexOf(':') + 1);
const stem = `${author.slice(author.lastIndexOf('.') + 1)}_${repo}`;
if (doc.type === 'file') {
const suffix =
engine === 'cli'
? path.extname(new URL(url).pathname)
: `.${url.slice(url.lastIndexOf('.') + 1)}`;
const raw = `${stem}${suffix}`;
return { script: `raw_${raw}`, raw };
}
const directory = doc.branch ? `${stem}_${doc.branch}` : stem;
return { script: directory, repo: directory };
}
// A subscription may have run through either engine before being deleted.
export function getSubscriptionStorageCandidates(doc: Subscription) {
const candidates: ReturnType<typeof getSubscriptionStorageNames>[] = [];
for (const engine of ['shell', 'cli'] as const) {
try {
candidates.push(getSubscriptionStorageNames(doc, engine));
} catch {
// Historical rows may have an invalid SSH alias or missing credentials.
// Retain their URL-derived storage without needing an executable SSH URL.
try {
candidates.push(
getSubscriptionStorageNames(
{
url: doc.url,
alias: doc.alias,
branch: doc.branch,
type: doc.type === 'private-repo' ? 'public-repo' : doc.type,
},
engine,
),
);
} catch {
// One engine's invalid URL must not hide the other engine's candidate.
}
}
}
return candidates;
}
+26
View File
@@ -0,0 +1,26 @@
import { QueryTypes, Sequelize } from 'sequelize';
export async function migrateUserSchema(database: Sequelize): Promise<void> {
const columns = [
{ table: 'Users', column: 'sessions', type: "JSON NOT NULL DEFAULT '[]'" },
...['Envs', 'Crontabs', 'Subscriptions', 'Dependences'].map((table) => ({
table,
column: 'userId',
type: 'INTEGER',
})),
];
await database.transaction(async (transaction) => {
for (const { table, column, type } of columns) {
const fields = await database.query<{ name: string }>(
`PRAGMA table_info("${table}")`,
{ type: QueryTypes.SELECT, transaction },
);
if (!fields.length)
throw new Error(`User migration table is missing: ${table}`);
if (!fields.some((field) => field.name === column))
await database.query(
`ALTER TABLE "${table}" ADD COLUMN "${column}" ${type}`,
{ transaction },
);
}
});
}
+1 -1
View File
@@ -4,7 +4,7 @@ import path from 'path';
import { writeFile, open, chmod } from 'fs/promises';
import { fileExist } from '../config/util';
function getUniqueLockPath(filePath: string) {
export function getUniqueLockPath(filePath: string) {
const sanitizedPath = filePath
.replace(/[<>:"/\\|?*]/g, '_')
.replace(/^_/, '');
+5 -16
View File
@@ -2,30 +2,19 @@ import 'reflect-metadata';
import OpenService from './services/open';
import { Container } from 'typedi';
import LoggerInstance from './loaders/logger';
import config from './config';
import path from 'path';
import os from 'os';
import { writeFileWithLock } from './shared/utils';
const tokenFile = path.join(config.configPath, 'token.json');
async function getToken() {
try {
Container.set('logger', LoggerInstance);
const openService = Container.get(OpenService);
const appToken = await openService.generateSystemToken();
const appToken = await openService.generateSystemToken(
process.argv.includes('--renew'),
);
console.log(appToken.value);
await writeFile({
value: appToken.value,
expiration: appToken.expiration,
});
} catch (error) {
console.log(error);
console.error(error);
process.exitCode = 1;
}
}
async function writeFile(data: any) {
await writeFileWithLock(tokenFile, `${JSON.stringify(data)}${os.EOL}`);
}
getToken();
+5
View File
@@ -6,6 +6,11 @@ declare global {
namespace Express {
interface Request {
platform: 'desktop' | 'mobile';
user?: {
userId?: number;
role?: number;
};
auth?: any;
}
}
}
+4 -20
View File
@@ -1,5 +1,5 @@
import { Joi } from 'celebrate';
import CronExpressionParser from 'cron-parser';
import { isValidCronSchedule } from '../shared/cronSchedule';
import { ScheduleType } from '../interface/schedule';
import path from 'path';
import config from '../config';
@@ -12,23 +12,7 @@ const validateSchedule = (value: string, helpers: any) => {
return value;
}
// 检测裸 /N 模式:cron-parser 会接受,但 node-schedule 会返回 null
// 提前拦截,避免任务入库后调度器注册失败
if (/\s\/\d/.test(value) || /^\/\d/.test(value)) {
return helpers.error('any.invalid');
}
// 检测 ? 字符:Quartz cron 语法,node-schedule 在大多数字段上返回 null
if (/\?/.test(value)) {
return helpers.error('any.invalid');
}
try {
if (CronExpressionParser.parse(value).hasNext()) {
return value;
}
} catch (e) {
return helpers.error('any.invalid');
}
if (isValidCronSchedule(value)) return value;
return helpers.error('any.invalid');
};
@@ -75,7 +59,7 @@ export const commonCronSchema = {
}
if (
!/^(?!.*(?:^|\/)\.{1,2}(?:\/|$))(?:\/)?(?:[\w.-]+\/)*[\w.-]+\/?$/.test(
!/^(?!.*(?:^|\/)\.{1,2}(?:\/|$))(?:\/)?(?:[\w\p{Script=Han}.-]+\/)*[\w\p{Script=Han}.-]+\/?$/u.test(
value,
)
) {
@@ -87,7 +71,7 @@ export const commonCronSchema = {
return value;
})
.messages({
'string.pattern.base': '日志名称只能包含字母、数字、下划线和连字符',
'string.pattern.base': '日志名称只能包含中文、字母、数字、下划线、连字符、点和路径分隔符',
'string.max': '日志名称不能超过100个字符',
'string.unsafePath': '绝对路径必须在日志目录内或使用 /dev/null',
}),
+2
View File
@@ -0,0 +1,2 @@
/dist/
node_modules
+201
View File
@@ -0,0 +1,201 @@
Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/
TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
1. Definitions.
"License" shall mean the terms and conditions for use, reproduction,
and distribution as defined by Sections 1 through 9 of this document.
"Licensor" shall mean the copyright owner or entity authorized by
the copyright owner that is granting the License.
"Legal Entity" shall mean the union of the acting entity and all
other entities that control, are controlled by, or are under common
control with that entity. For the purposes of this definition,
"control" means (i) the power, direct or indirect, to cause the
direction or management of such entity, whether by contract or
otherwise, or (ii) ownership of fifty percent (50%) or more of the
outstanding shares, or (iii) beneficial ownership of such entity.
"You" (or "Your") shall mean an individual or Legal Entity
exercising permissions granted by this License.
"Source" form shall mean the preferred form for making modifications,
including but not limited to software source code, documentation
source, and configuration files.
"Object" form shall mean any form resulting from mechanical
transformation or translation of a Source form, including but
not limited to compiled object code, generated documentation,
and conversions to other media types.
"Work" shall mean the work of authorship, whether in Source or
Object form, made available under the License, as indicated by a
copyright notice that is included in or attached to the work
(an example is provided in the Appendix below).
"Derivative Works" shall mean any work, whether in Source or Object
form, that is based on (or derived from) the Work and for which the
editorial revisions, annotations, elaborations, or other modifications
represent, as a whole, an original work of authorship. For the purposes
of this License, Derivative Works shall not include works that remain
separable from, or merely link (or bind by name) to the interfaces of,
the Work and Derivative Works thereof.
"Contribution" shall mean any work of authorship, including
the original version of the Work and any modifications or additions
to that Work or Derivative Works thereof, that is intentionally
submitted to Licensor for inclusion in the Work by the copyright owner
or by an individual or Legal Entity authorized to submit on behalf of
the copyright owner. For the purposes of this definition, "submitted"
means any form of electronic, verbal, or written communication sent
to the Licensor or its representatives, including but not limited to
communication on electronic mailing lists, source code control systems,
and issue tracking systems that are managed by, or on behalf of, the
Licensor for the purpose of discussing and improving the Work, but
excluding communication that is conspicuously marked or otherwise
designated in writing by the copyright owner as "Not a Contribution."
"Contributor" shall mean Licensor and any individual or Legal Entity
on behalf of whom a Contribution has been received by Licensor and
subsequently incorporated within the Work.
2. Grant of Copyright License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
copyright license to reproduce, prepare Derivative Works of,
publicly display, publicly perform, sublicense, and distribute the
Work and such Derivative Works in Source or Object form.
3. Grant of Patent License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
(except as stated in this section) patent license to make, have made,
use, offer to sell, sell, import, and otherwise transfer the Work,
where such license applies only to those patent claims licensable
by such Contributor that are necessarily infringed by their
Contribution(s) alone or by combination of their Contribution(s)
with the Work to which such Contribution(s) was submitted. If You
institute patent litigation against any entity (including a
cross-claim or counterclaim in a lawsuit) alleging that the Work
or a Contribution incorporated within the Work constitutes direct
or contributory patent infringement, then any patent licenses
granted to You under this License for that Work shall terminate
as of the date such litigation is filed.
4. Redistribution. You may reproduce and distribute copies of the
Work or Derivative Works thereof in any medium, with or without
modifications, and in Source or Object form, provided that You
meet the following conditions:
(a) You must give any other recipients of the Work or
Derivative Works a copy of this License; and
(b) You must cause any modified files to carry prominent notices
stating that You changed the files; and
(c) You must retain, in the Source form of any Derivative Works
that You distribute, all copyright, patent, trademark, and
attribution notices from the Source form of the Work,
excluding those notices that do not pertain to any part of
the Derivative Works; and
(d) If the Work includes a "NOTICE" text file as part of its
distribution, then any Derivative Works that You distribute must
include a readable copy of the attribution notices contained
within such NOTICE file, excluding those notices that do not
pertain to any part of the Derivative Works, in at least one
of the following places: within a NOTICE text file distributed
as part of the Derivative Works; within the Source form or
documentation, if provided along with the Derivative Works; or,
within a display generated by the Derivative Works, if and
wherever such third-party notices normally appear. The contents
of the NOTICE file are for informational purposes only and
do not modify the License. You may add Your own attribution
notices within Derivative Works that You distribute, alongside
or as an addendum to the NOTICE text from the Work, provided
that such additional attribution notices cannot be construed
as modifying the License.
You may add Your own copyright statement to Your modifications and
may provide additional or different license terms and conditions
for use, reproduction, or distribution of Your modifications, or
for any such Derivative Works as a whole, provided Your use,
reproduction, and distribution of the Work otherwise complies with
the conditions stated in this License.
5. Submission of Contributions. Unless You explicitly state otherwise,
any Contribution intentionally submitted for inclusion in the Work
by You to the Licensor shall be under the terms and conditions of
this License, without any additional terms or conditions.
Notwithstanding the above, nothing herein shall supersede or modify
the terms of any separate license agreement you may have executed
with Licensor regarding such Contributions.
6. Trademarks. This License does not grant permission to use the trade
names, trademarks, service marks, or product names of the Licensor,
except as required for reasonable and customary use in describing the
origin of the Work and reproducing the content of the NOTICE file.
7. Disclaimer of Warranty. Unless required by applicable law or
agreed to in writing, Licensor provides the Work (and each
Contributor provides its Contributions) on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
implied, including, without limitation, any warranties or conditions
of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
PARTICULAR PURPOSE. You are solely responsible for determining the
appropriateness of using or redistributing the Work and assume any
risks associated with Your exercise of permissions under this License.
8. Limitation of Liability. In no event and under no legal theory,
whether in tort (including negligence), contract, or otherwise,
unless required by applicable law (such as deliberate and grossly
negligent acts) or agreed to in writing, shall any Contributor be
liable to You for damages, including any direct, indirect, special,
incidental, or consequential damages of any character arising as a
result of this License or out of the use or inability to use the
Work (including but not limited to damages for loss of goodwill,
work stoppage, computer failure or malfunction, or any and all
other commercial damages or losses), even if such Contributor
has been advised of the possibility of such damages.
9. Accepting Warranty or Additional Liability. While redistributing
the Work or Derivative Works thereof, You may choose to offer,
and charge a fee for, acceptance of support, warranty, indemnity,
or other liability obligations and/or rights consistent with this
License. However, in accepting such obligations, You may act only
on Your own behalf and on Your sole responsibility, not on behalf
of any other Contributor, and only if You agree to indemnify,
defend, and hold each Contributor harmless for any liability
incurred by, or claims asserted against, such Contributor by reason
of your accepting any such warranty or additional liability.
END OF TERMS AND CONDITIONS
APPENDIX: How to apply the Apache License to your work.
To apply the Apache License to your work, attach the following
boilerplate notice, with the fields enclosed by brackets "{}"
replaced with your own identifying information. (Don't include
the brackets!) The text should be enclosed in the appropriate
comment syntax for the file format. We also recommend that a
file or class name and description of purpose be included on the
same "printed page" as the copyright notice for easier
identification within third-party archives.
Copyright 2021 WHYOUR <https://github.com/whyour>.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
+45
View File
@@ -0,0 +1,45 @@
# Panel-internal TypeScript tools
[简体中文](LOCAL.md) | **English**
The npm and panel-internal entries both use the name `ql`, but have separate Commander command trees. The npm entry only calls remote APIs; the internal entry only runs local tools. Verify the absolute executable path and `--help` before use. Installing the npm package does not migrate the built-in Shell commands.
These tools ship with panel builds and are excluded from the @whyour/qinglong-cli npm package. Node >=22.12 is required; user configuration/hooks remain Bash and scripts need their interpreters. Development publishing was removed from the CLI; shell/pub.sh remains available to release workflows.
```sh
npm ci --prefix cli
npm run build:cli
node cli/dist/ql.js --help
node cli/dist/ql.js task exec --root /ql demo.js now
node cli/dist/ql.js reload --root /ql
```
Run on the actual panel host or inside its container. For Docker use `docker exec <container> <selected-absolute-entry> ...`. A workstation npm CLI cannot reset a remote account; a mounted data directory alone is not the complete operational environment.
| Capability | Internal commands |
| --- | --- |
| Execution/inventory | `ql task exec [options] [script]`, with `ql task <script>` and `task <script>` shorthands |
| Subscription synchronization | `ql repo <url> [include] [exclude] [dependencies] [branch] [extensions] [proxy] [autoAdd] [autoDelete]`, `ql raw <url> [proxy] [autoAdd] [autoDelete]` |
| Startup/repair | `ql start`, `ql repair-config`, `ql check` |
| Upgrade/reload | `ql update [--mirror github\|gitee] [--download-only]`, `ql reload [--target services\|system\|data]` |
| Logs/extensions | `ql rmlog <days>`, `ql extra`, `ql bot` |
| Recovery | `ql resetlet`, `ql resettfa`, `ql resetpwd -- <value>`, `ql resetname -- <value>` |
`ql local <command>` remains compatible. `update false` means --download-only; reload system/data/services maps to --target. Place maintenance --root, --data-dir and --json before --. Password values appear in process arguments: use the owner's trusted terminal and never request passwords in chat.
Runner options precede the script. now skips delay; conc/desi select accounts; -- passes subsequent arguments through. With QL_DIR set, bare ql task or task lists JS scripts. Explicit --help does not read configuration. JSON mode sends script output to stderr and the final result to stdout, retaining the script exit code. The internal entry rejects remote API commands and does not read remote credentials. Use the separate npm entry for remote management; use task exec for scripts with reserved API action names.
repo/raw retain positional arguments and QL_DIR/QL_DATA_DIR, without --root/--json. Filtering uses native grep -E POSIX ERE; Git/curl and other operation-specific tools are required. The configuration bridge uses Bash and env supporting -0.
check installs dependencies, repairs and reloads; inspect its before/after observations rather than treating it as a read-only probe. reload defaults to service restart; system/data apply staged files. start --no-startup skips OS boot registration for containers; start --reload skips installation, optional hooks and boot registration. Bot setup involves system/pip dependencies and external processes.
## Integration and skill
On a panel containing the selection loader, set `QL_CLI_ROOT=/absolute/path/to/built/cli` to the internal build directory containing the full dist tree, **not an npm installation**. Restart to create private ~/bin ql/task/cron wrappers and prioritize them for panel processes. Independent terminals should use the selected absolute path. Unset the variable and restart to restore Shell entries. Invalid paths report errors without silent fallback. Original Shell entries remain the default.
dist/startup.js retains startup/reload compatibility; dist/compat.js and dist/subscription-worker.js are internal adapters, not npm commands. Lifecycle fields follow the installed panel version; customized versions may select QL_CLI_LIFECYCLE=legacy|extended.
The separate [qinglong-local skill](skills/qinglong-local/SKILL.md) covers execution and maintenance. Use qinglong-cli for remote API management. QL_LANG=en selects English help/messages while JSON fields remain stable.
Distinguish isolated fixtures, real-panel validation, and actual system upgrades/reboots. Fixture success does not establish support for every distribution, Bot or public upgrade download. Development publishing is explicitly excluded. Performance comparisons use temporary evaluation scripts, not shipped product code.
+47
View File
@@ -0,0 +1,47 @@
# 面板内部 TypeScript 工具
**简体中文** | [English](LOCAL.en.md)
npm 与面板内部入口都叫 `ql`,但使用独立的 Commander 命令树:npm 入口只调用远程 API,内部入口只运行本机工具。使用前确认可执行文件的绝对路径和 `--help`;安装 npm 包不会迁移内置 Shell 命令。
这些工具随面板构建交付,不包含在 `@whyour/qinglong-cli` npm 包中。要求 Node >=22.12;用户配置和 hook 保持 Bash,任务还需要对应解释器。开发发布命令已移除,原 `shell/pub.sh` 保留供发布流程使用。
```sh
npm ci --prefix cli
npm run build:cli
node cli/dist/ql.js --help
node cli/dist/ql.js task exec --root /ql demo.js now
node cli/dist/ql.js reload --root /ql
```
必须在实际安装面板的宿主机或容器中运行。Docker 使用 `docker exec <容器> <容器内选定入口> ...`;不能在工作站上用 npm ql 重置远端账号,单独挂载 data 目录也不是完整运行环境。
## 命令
| 能力 | 内部命令 |
| --- | --- |
| 脚本执行/清单 | `ql task exec [选项] [脚本]`,兼容 `ql task <脚本>` 和 `task <脚本>` |
| 订阅同步 | `ql repo <url> [include] [exclude] [dependencies] [branch] [extensions] [proxy] [autoAdd] [autoDelete]`、`ql raw <url> [proxy] [autoAdd] [autoDelete]` |
| 启动与修复 | `ql start`、`ql repair-config`、`ql check` |
| 升级/重载 | `ql update [--mirror github\|gitee] [--download-only]`、`ql reload [--target services\|system\|data]` |
| 日志与扩展 | `ql rmlog <days>`、`ql extra`、`ql bot` |
| 账号恢复 | `ql resetlet`、`ql resettfa`、`ql resetpwd -- <value>`、`ql resetname -- <value>` |
`ql local <命令>` 保留为兼容写法。`update false` 等价于 `--download-only`,`reload system/data/services` 映射到 `--target`。维护选项 `--root`、`--data-dir`、`--json` 必须放在 `--` 前。密码位置参数会出现在进程参数中,使用本人可信终端,不向聊天发送密码。
执行器选项必须放在脚本前;`now` 跳过延迟,`conc`/`desi` 支持账号选择,`--` 后参数透传。设置 QL_DIR 后,无参数 `ql task` 或 `task` 显示 JS 脚本清单;显式 `--help` 不读取配置。`--json` 将脚本输出写 stderr、最终结果写 stdout,并保留脚本退出码。内部入口拒绝远程 API 命令,不读取远程认证配置。远程管理使用独立 npm 入口;与 API 动作同名的脚本使用显式 `task exec`。
repo/raw 沿用位置参数,使用 QL_DIR/QL_DATA_DIR,不接受 --root/--json;筛选采用本机 `grep -E` 的 POSIX ERE,需 Git/curl 等系统工具。配置桥使用 Bash 和支持 `-0` 的 env。
`check` 会安装依赖、修复并重载,并非只读探测;结果须检查 before/after 健康观测。`reload` 默认重启服务,system/data 应用暂存文件。`start --no-startup` 跳过 OS 启动注册,适用于容器;`start --reload` 跳过依赖安装、可选 hook 与启动注册。Bot 涉及系统/pip 依赖与外部进程。
## 集成与 Skill
包含选择加载器的面板设置 `QL_CLI_ROOT=/absolute/path/to/built/cli`,指向包含完整 dist 的面板工具构建目录,**不能指向 npm 包**。重启后加载器在 ~/bin 创建私有 ql/task/cron 包装器,并为面板进程优先使用它们;独立终端应显式使用该路径。清除变量并重启恢复原 Shell 入口。无效路径报错,不静默降级。默认仍保留旧 Shell 入口。
`dist/startup.js` 兼容旧启动/reload;`dist/compat.js` 与 `dist/subscription-worker.js` 保留内部适配用途,不注册为 npm 命令。生命周期根据已安装面板版本选择字段,定制版本可设置 QL_CLI_LIFECYCLE=legacy|extended。
独立 Skill 位于 [skills/qinglong-local](skills/qinglong-local/SKILL.md),包含完整执行与维护参考。远程管理另用 qinglong-cli Skill。`QL_LANG=en` 切换帮助和执行提示;JSON 字段保持一致。
功能验收区分隔离夹具、真实面板和真实系统升级/重启。不能将夹具通过视为所有发行版、Bot 或公网升级均已验证。开发发布从迁移目标中明确排除;性能比较使用临时评测脚本,不随产品发布。

Some files were not shown because too many files have changed in this diff Show More