diff --git a/.gitignore b/.gitignore index b69ea28..44d4d55 100644 --- a/.gitignore +++ b/.gitignore @@ -4,12 +4,21 @@ __pycache__/ *.py[cod] .pytest_cache/ +# Runtime logs +/logs/ +*.log + # Local docs and agent instructions -/docs/ +/docs/* +!/docs/ai_integration.md +!/docs/mysql_integration.md /read.md -/README.md /AGENTS.md /AGENTS.md.bak-* # Local migration workspace /migration/ + +# Local generated docs +/docs/ +/README.md diff --git a/Dockerfile b/Dockerfile index e3e1101..0982cf4 100644 --- a/Dockerfile +++ b/Dockerfile @@ -14,6 +14,7 @@ RUN pip install --no-cache-dir \ "psycopg[binary]==3.2.3" \ pydantic-settings==2.7.1 \ python-dotenv==1.0.1 \ + alembic==1.14.0 \ httpx==0.28.1 \ apscheduler==3.10.4 \ redis==5.2.1 \ diff --git a/alembic/env.py b/alembic/env.py index f9be3b1..3ab7599 100644 --- a/alembic/env.py +++ b/alembic/env.py @@ -4,7 +4,7 @@ from alembic import context from sqlalchemy import engine_from_config, pool from app.core.config import get_settings -from app.core.database import Base +from app.core.db_base import Base from app.modules.approvals import models as approval_models from app.modules.audit import models as audit_models from app.modules.business import models as business_models diff --git a/alembic/versions/202607060001_initial_schema.py b/alembic/versions/202607060001_initial_schema.py index efbf7ba..8b1969a 100644 --- a/alembic/versions/202607060001_initial_schema.py +++ b/alembic/versions/202607060001_initial_schema.py @@ -7,7 +7,7 @@ Create Date: 2026-07-06 from alembic import op -from app.core.database import Base +from app.core.db_base import Base from app.modules.approvals import models as approval_models from app.modules.audit import models as audit_models from app.modules.business import models as business_models @@ -26,4 +26,4 @@ def upgrade() -> None: def downgrade() -> None: - Base.metadata.drop_all(bind=op.get_bind()) + raise RuntimeError("Initial schema downgrade is destructive; create an explicit rollback plan.") diff --git a/alembic/versions/202607060002_approval_ticket_consumption.py b/alembic/versions/202607060002_approval_ticket_consumption.py new file mode 100644 index 0000000..a729b16 --- /dev/null +++ b/alembic/versions/202607060002_approval_ticket_consumption.py @@ -0,0 +1,57 @@ +"""Add approval ticket consumption fields. + +Revision ID: 202607060002 +Revises: 202607060001 +Create Date: 2026-07-06 +""" + +from alembic import op +import sqlalchemy as sa +from sqlalchemy import inspect + + +revision = "202607060002" +down_revision = "202607060001" +branch_labels = None +depends_on = None + +APPROVAL_REQUESTS_TABLE = "approval_requests" +USED_BY_COLUMN = "used_by" +USED_AT_COLUMN = "used_at" + + +def _column_names() -> set[str]: + inspector = inspect(op.get_bind()) + return { + column["name"] + for column in inspector.get_columns(APPROVAL_REQUESTS_TABLE) + } + + +def upgrade() -> None: + columns = _column_names() + if USED_BY_COLUMN not in columns: + op.add_column( + APPROVAL_REQUESTS_TABLE, + sa.Column(USED_BY_COLUMN, sa.String(length=128), nullable=True), + ) + op.create_index( + op.f("ix_approval_requests_used_by"), + APPROVAL_REQUESTS_TABLE, + [USED_BY_COLUMN], + unique=False, + ) + if USED_AT_COLUMN not in columns: + op.add_column( + APPROVAL_REQUESTS_TABLE, + sa.Column(USED_AT_COLUMN, sa.DateTime(), nullable=True), + ) + + +def downgrade() -> None: + columns = _column_names() + if USED_BY_COLUMN in columns: + op.drop_index(op.f("ix_approval_requests_used_by"), table_name=APPROVAL_REQUESTS_TABLE) + op.drop_column(APPROVAL_REQUESTS_TABLE, USED_BY_COLUMN) + if USED_AT_COLUMN in columns: + op.drop_column(APPROVAL_REQUESTS_TABLE, USED_AT_COLUMN) diff --git a/app/core/config.py b/app/core/config.py index c47c22e..93d6460 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -1,4 +1,6 @@ +import json from functools import lru_cache +from typing import Any from pydantic import Field, field_validator from pydantic_settings import BaseSettings, SettingsConfigDict @@ -19,11 +21,14 @@ class Settings(BaseSettings): api_prefix: str = "/api/v1" api_key: str | None = None api_actor: str = ActorValue.API + approval_api_key: str | None = None + approval_api_actor: str = ActorValue.APPROVER cors_origins: list[str] = Field(default_factory=lambda: ["*"]) database_url: str = "mysql+pymysql://root:password@127.0.0.1:3306/company_ai?charset=utf8mb4" legacy_database_url: str | None = None legacy_project_query: str | None = None + legacy_allowed_queries: dict[str, str] = Field(default_factory=dict) legacy_project_code_prefix: str = "LEGACY" redis_url: str = "redis://127.0.0.1:6379/0" @@ -71,6 +76,20 @@ class Settings(BaseSettings): return value return [item.strip() for item in value.split(",") if item.strip()] + @field_validator("legacy_allowed_queries", mode="before") + @classmethod + def parse_legacy_allowed_queries(cls, value: Any) -> dict[str, str]: + if value is None or value == "": + return {} + if isinstance(value, dict): + return {str(key): str(item) for key, item in value.items()} + if isinstance(value, str): + data = json.loads(value) + if not isinstance(data, dict): + raise ValueError("LEGACY_ALLOWED_QUERIES must be a JSON object") + return {str(key): str(item) for key, item in data.items()} + raise ValueError("LEGACY_ALLOWED_QUERIES must be a JSON object") + @lru_cache def get_settings() -> Settings: diff --git a/app/core/constants.py b/app/core/constants.py index 4396b32..35fa75e 100644 --- a/app/core/constants.py +++ b/app/core/constants.py @@ -3,6 +3,7 @@ from enum import StrEnum class ActorValue(StrEnum): API = "api" + APPROVER = "approver" SYSTEM = "system" SCHEDULER = "scheduler" FEISHU = "feishu" @@ -10,6 +11,8 @@ class ActorValue(StrEnum): class HttpHeader(StrEnum): AUTHORIZATION = "Authorization" + X_API_KEY = "X-API-Key" + X_APPROVAL_API_KEY = "X-Approval-API-Key" BEARER_TOKEN_TEMPLATE = "Bearer {token}" diff --git a/app/core/database.py b/app/core/database.py index e8e78bb..2e41d9b 100644 --- a/app/core/database.py +++ b/app/core/database.py @@ -1,15 +1,10 @@ from collections.abc import Generator from sqlalchemy import create_engine -from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker +from sqlalchemy.orm import Session, sessionmaker from app.core.config import get_settings - - -class Base(DeclarativeBase): - """Base class for SQLAlchemy ORM models.""" - - pass +from app.core.db_base import Base settings = get_settings() diff --git a/app/core/db_base.py b/app/core/db_base.py new file mode 100644 index 0000000..79161fd --- /dev/null +++ b/app/core/db_base.py @@ -0,0 +1,7 @@ +from sqlalchemy.orm import DeclarativeBase + + +class Base(DeclarativeBase): + """Base class for SQLAlchemy ORM models.""" + + pass diff --git a/app/core/security.py b/app/core/security.py index 6c9e432..2ade8e6 100644 --- a/app/core/security.py +++ b/app/core/security.py @@ -1,8 +1,10 @@ from dataclasses import dataclass +from secrets import compare_digest from fastapi import Header, HTTPException, status from app.core.config import get_settings +from app.core.constants import HttpHeader @dataclass(frozen=True) @@ -12,7 +14,9 @@ class ApiPrincipal: actor: str -def require_api_key(x_api_key: str | None = Header(default=None)) -> ApiPrincipal: +def require_api_key( + x_api_key: str | None = Header(default=None, alias=HttpHeader.X_API_KEY), +) -> ApiPrincipal: """Validate the internal API key header and return its service principal.""" settings = get_settings() @@ -21,6 +25,31 @@ def require_api_key(x_api_key: str | None = Header(default=None)) -> ApiPrincipa status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="API_KEY is required", ) - if x_api_key != settings.api_key: + if not x_api_key or not compare_digest(x_api_key, settings.api_key): raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid API key") return ApiPrincipal(actor=settings.api_actor) + + +def require_approval_api_key( + x_approval_api_key: str | None = Header( + default=None, + alias=HttpHeader.X_APPROVAL_API_KEY, + ), +) -> ApiPrincipal: + """Validate the approval API key and return the approval principal.""" + + settings = get_settings() + if not settings.approval_api_key: + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail="APPROVAL_API_KEY is required", + ) + if ( + not x_approval_api_key + or not compare_digest(x_approval_api_key, settings.approval_api_key) + ): + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Invalid approval API key", + ) + return ApiPrincipal(actor=settings.approval_api_actor) diff --git a/app/modules/ai_agent/adapters.py b/app/modules/ai_agent/adapters.py index b5bebbe..438f00e 100644 --- a/app/modules/ai_agent/adapters.py +++ b/app/modules/ai_agent/adapters.py @@ -263,8 +263,13 @@ class OpenClawHermesAdapter(AIAdapter): or AIDefault.SESSION_KEY_MAIN ), ) + except HTTPException: + raise except Exception as exc: - result[AIResponseKey.TOOL_ERROR] = _error_detail(exc) + raise HTTPException( + status_code=502, + detail={AIErrorKey.OPENCLAW: _error_detail(exc)}, + ) from exc return result def _recall_memory(self, prompt: str, context: dict[str, Any]) -> dict[str, Any]: diff --git a/app/modules/approvals/constants.py b/app/modules/approvals/constants.py index 21e1f8f..61c46cc 100644 --- a/app/modules/approvals/constants.py +++ b/app/modules/approvals/constants.py @@ -5,8 +5,17 @@ class ApprovalStatus(StrEnum): PENDING = "pending" APPROVED = "approved" REJECTED = "rejected" + USED = "used" class ApprovalActionValue(StrEnum): + CREATE = "create" UPDATE = "update" - WILDCARD = "*" + + +class ApprovalErrorDetail(StrEnum): + NOT_FOUND = "Approval ticket not found" + ALREADY_DECIDED = "Approval ticket already decided" + SELF_APPROVAL = "Approval applicant cannot approve their own ticket" + NOT_APPROVED = "Approval ticket is not approved for this change" + PAYLOAD_MISMATCH = "Approval ticket payload does not match this change" diff --git a/app/modules/approvals/models.py b/app/modules/approvals/models.py index 42d8769..9fb7c80 100644 --- a/app/modules/approvals/models.py +++ b/app/modules/approvals/models.py @@ -4,7 +4,7 @@ from sqlalchemy import DateTime, Integer, String, Text from sqlalchemy.orm import Mapped, mapped_column from app.core.constants import ActorValue -from app.core.database import Base +from app.core.db_base import Base from app.core.time import utc_now from app.modules.approvals.constants import ApprovalStatus @@ -23,6 +23,7 @@ class ApprovalRequest(Base): reason: Mapped[str | None] = mapped_column(Text, nullable=True) payload: Mapped[str | None] = mapped_column(Text, nullable=True) decision_comment: Mapped[str | None] = mapped_column(Text, nullable=True) + used_by: Mapped[str | None] = mapped_column(String(128), nullable=True, index=True) created_at: Mapped[datetime] = mapped_column(DateTime, default=utc_now, index=True) updated_at: Mapped[datetime] = mapped_column( DateTime, @@ -30,3 +31,4 @@ class ApprovalRequest(Base): onupdate=utc_now, ) decided_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + used_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) diff --git a/app/modules/approvals/routes.py b/app/modules/approvals/routes.py index a010616..99c9104 100644 --- a/app/modules/approvals/routes.py +++ b/app/modules/approvals/routes.py @@ -2,7 +2,7 @@ from fastapi import APIRouter, Depends from sqlalchemy.orm import Session from app.core.database import get_db -from app.core.security import ApiPrincipal, require_api_key +from app.core.security import ApiPrincipal, require_api_key, require_approval_api_key from app.modules.approvals.schemas import ApprovalCreate, ApprovalDecision, ApprovalRead from app.modules.approvals.service import ApprovalService @@ -37,7 +37,7 @@ def approve( ticket_id: str, payload: ApprovalDecision, db: Session = Depends(get_db), - principal: ApiPrincipal = Depends(require_api_key), + principal: ApiPrincipal = Depends(require_approval_api_key), ): return ApprovalService(db).decide(ticket_id, principal.actor, True, payload.comment) @@ -47,6 +47,6 @@ def reject( ticket_id: str, payload: ApprovalDecision, db: Session = Depends(get_db), - principal: ApiPrincipal = Depends(require_api_key), + principal: ApiPrincipal = Depends(require_approval_api_key), ): return ApprovalService(db).decide(ticket_id, principal.actor, False, payload.comment) diff --git a/app/modules/approvals/schemas.py b/app/modules/approvals/schemas.py index 1e6a028..f1f8988 100644 --- a/app/modules/approvals/schemas.py +++ b/app/modules/approvals/schemas.py @@ -34,6 +34,8 @@ class ApprovalRead(BaseModel): reason: str | None payload: str | None decision_comment: str | None + used_by: str | None created_at: datetime updated_at: datetime decided_at: datetime | None + used_at: datetime | None diff --git a/app/modules/approvals/service.py b/app/modules/approvals/service.py index e1051f0..f52f48e 100644 --- a/app/modules/approvals/service.py +++ b/app/modules/approvals/service.py @@ -1,12 +1,19 @@ import json import uuid +from datetime import date, datetime +from decimal import Decimal +from typing import Any from fastapi import HTTPException, status from sqlalchemy import select from sqlalchemy.orm import Session from app.core.time import utc_now -from app.modules.approvals.constants import ApprovalActionValue, ApprovalStatus +from app.modules.approvals.constants import ( + ApprovalActionValue, + ApprovalErrorDetail, + ApprovalStatus, +) from app.modules.approvals.models import ApprovalRequest from app.modules.approvals.schemas import ApprovalCreate from app.modules.audit.constants import AuditAction, AuditRiskLevel, AuditSource @@ -61,7 +68,7 @@ class ApprovalService: if ticket is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, - detail="Approval ticket not found", + detail=ApprovalErrorDetail.NOT_FOUND, ) return ticket @@ -74,7 +81,15 @@ class ApprovalService: ) -> ApprovalRequest: ticket = self.get_by_ticket(ticket_id) if ticket.status != ApprovalStatus.PENDING: - raise HTTPException(status_code=409, detail="Approval ticket already decided") + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=ApprovalErrorDetail.ALREADY_DECIDED, + ) + if approved and approver == ticket.applicant: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=ApprovalErrorDetail.SELF_APPROVAL, + ) ticket.status = ApprovalStatus.APPROVED if approved else ApprovalStatus.REJECTED ticket.approver = approver ticket.decision_comment = comment @@ -95,6 +110,33 @@ class ApprovalService: ) return ticket + def consume_for( + self, + ticket_id: str, + domain: str, + record_id: str | int | None, + action: str, + payload: dict[str, Any], + actor: str, + ) -> ApprovalRequest: + ticket = self.get_by_ticket(ticket_id) + if not self._is_ticket_scope_valid(ticket, domain, record_id, action): + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=ApprovalErrorDetail.NOT_APPROVED, + ) + if not _payload_matches(ticket.payload, payload): + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=ApprovalErrorDetail.PAYLOAD_MISMATCH, + ) + ticket.status = ApprovalStatus.USED + ticket.used_by = actor + ticket.used_at = utc_now() + if record_id is not None and not ticket.record_id: + ticket.record_id = str(record_id) + return ticket + def is_approved_for( self, ticket_id: str, @@ -103,6 +145,15 @@ class ApprovalService: action: str, ) -> bool: ticket = self.get_by_ticket(ticket_id) + return self._is_ticket_scope_valid(ticket, domain, record_id, action) + + @staticmethod + def _is_ticket_scope_valid( + ticket: ApprovalRequest, + domain: str, + record_id: str | int | None, + action: str, + ) -> bool: if ticket.status != ApprovalStatus.APPROVED: return False if ticket.domain != domain: @@ -115,5 +166,28 @@ class ApprovalService: action, ApprovalActionValue.UPDATE, f"{ApprovalActionValue.UPDATE}:{domain}", - ApprovalActionValue.WILDCARD, } + + +def _payload_matches(approved_payload: str | None, requested_payload: dict[str, Any]) -> bool: + try: + parsed_payload = json.loads(approved_payload or "{}") + except json.JSONDecodeError: + parsed_payload = {} + return _canonical_payload(parsed_payload) == _canonical_payload(requested_payload) + + +def _canonical_payload(value: Any) -> str: + return json.dumps(_json_safe(value), ensure_ascii=False, sort_keys=True, default=str) + + +def _json_safe(value: Any) -> Any: + if isinstance(value, Decimal): + return float(value) + if isinstance(value, (datetime, date)): + return value.isoformat() + if isinstance(value, dict): + return {str(key): _json_safe(item) for key, item in value.items()} + if isinstance(value, list): + return [_json_safe(item) for item in value] + return value diff --git a/app/modules/audit/models.py b/app/modules/audit/models.py index 8e171ef..32a412c 100644 --- a/app/modules/audit/models.py +++ b/app/modules/audit/models.py @@ -4,7 +4,7 @@ from sqlalchemy import DateTime, Integer, String, Text from sqlalchemy.orm import Mapped, mapped_column from app.core.constants import ActorValue -from app.core.database import Base +from app.core.db_base import Base from app.core.time import utc_now from app.modules.audit.constants import AuditRiskLevel, AuditSource, AuditStatus diff --git a/app/modules/business/models.py b/app/modules/business/models.py index d5788d7..939d270 100644 --- a/app/modules/business/models.py +++ b/app/modules/business/models.py @@ -5,7 +5,7 @@ from sqlalchemy import JSON, Date, DateTime, Integer, Numeric, String, Text from sqlalchemy.orm import Mapped, mapped_column from app.core.constants import ActorValue -from app.core.database import Base +from app.core.db_base import Base from app.core.time import utc_now from app.modules.business.constants import ( AccountType, diff --git a/app/modules/business/service.py b/app/modules/business/service.py index 395f343..ed3ada9 100644 --- a/app/modules/business/service.py +++ b/app/modules/business/service.py @@ -48,14 +48,17 @@ def _coerce_column_value(column: Column, value: Any) -> Any: def _model_payload(model: Any, data: dict[str, Any]) -> dict[str, Any]: - """Filter unknown keys and coerce values according to model column types.""" + """Validate keys and coerce values according to model column types.""" columns = {column.name: column for column in model.__table__.columns if column.name != "id"} payload: dict[str, Any] = {} for key, value in data.items(): column = columns.get(key) if column is None: - continue + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail=f"Unknown field '{key}'", + ) try: payload[key] = _coerce_column_value(column, value) except (ValueError, TypeError, InvalidOperation) as exc: @@ -104,12 +107,20 @@ class BusinessService: actor: str = ActorValue.API, approval_ticket_id: str | None = None, ) -> dict[str, Any]: - if domain in HIGH_RISK_DOMAINS: - self._ensure_approved(approval_ticket_id, domain, None, f"create:{domain}") model = get_domain_model(domain) payload = _model_payload(model, data) record = model(**payload) self.db.add(record) + if domain in HIGH_RISK_DOMAINS: + self.db.flush() + self._consume_approval( + approval_ticket_id, + domain, + record.id, + f"create:{domain}", + data, + actor, + ) self.db.commit() self.db.refresh(record) result = serialize_model(record) @@ -137,8 +148,6 @@ class BusinessService: actor: str = ActorValue.API, approval_ticket_id: str | None = None, ) -> dict[str, Any]: - if domain in HIGH_RISK_DOMAINS: - self._ensure_approved(approval_ticket_id, domain, record_id, f"update:{domain}") model = get_domain_model(domain) record = self.db.get(model, record_id) if record is None: @@ -146,7 +155,17 @@ class BusinessService: status_code=status.HTTP_404_NOT_FOUND, detail="Record not found", ) - for key, value in _model_payload(model, data).items(): + payload = _model_payload(model, data) + if domain in HIGH_RISK_DOMAINS: + self._consume_approval( + approval_ticket_id, + domain, + record_id, + f"update:{domain}", + data, + actor, + ) + for key, value in payload.items(): setattr(record, key, value) self.db.commit() self.db.refresh(record) @@ -167,25 +186,25 @@ class BusinessService: ) return result - def _ensure_approved( + def _consume_approval( self, approval_ticket_id: str | None, domain: str, record_id: str | int | None, action: str, + payload: dict[str, Any], + actor: str, ) -> None: if not approval_ticket_id: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail="High-risk domain change requires approval_ticket_id", ) - if not ApprovalService(self.db).is_approved_for( + ApprovalService(self.db).consume_for( approval_ticket_id, domain, record_id, action, - ): - raise HTTPException( - status_code=status.HTTP_403_FORBIDDEN, - detail="Approval ticket is not approved for this change", - ) + payload, + actor, + ) diff --git a/app/modules/legacy_mysql/constants.py b/app/modules/legacy_mysql/constants.py new file mode 100644 index 0000000..38134b9 --- /dev/null +++ b/app/modules/legacy_mysql/constants.py @@ -0,0 +1,10 @@ +from enum import StrEnum + + +class LegacyQueryName(StrEnum): + PROJECTS = "projects" + + +class LegacyQueryError(StrEnum): + QUERY_NOT_ALLOWED = "Legacy query is not in the configured allowlist" + PROJECT_QUERY_NOT_CONFIGURED = "LEGACY_PROJECT_QUERY is not configured. Configure it first." diff --git a/app/modules/legacy_mysql/routes.py b/app/modules/legacy_mysql/routes.py index 59d5454..5575b18 100644 --- a/app/modules/legacy_mysql/routes.py +++ b/app/modules/legacy_mysql/routes.py @@ -31,7 +31,10 @@ def describe_table(table_name: str, db: Session = Depends(get_db)) -> dict: @router.post("/query", response_model=QueryResult) def readonly_query(payload: ReadonlyQueryRequest, db: Session = Depends(get_db)) -> dict: - return LegacyMySQLService(db).execute_readonly(payload.sql, payload.params, payload.limit) + service = LegacyMySQLService(db) + if payload.sql: + return service.execute_readonly(payload.sql, payload.params, payload.limit) + return service.execute_allowed_query(payload.query_name, payload.params, payload.limit) @router.get("/projects", response_model=QueryResult) @@ -47,6 +50,7 @@ def sync_projects( ) -> dict: return LegacyMySQLService(db).sync_projects( source_query=payload.source_query, + source_query_name=payload.source_query_name, field_map=payload.field_map, limit=payload.limit, dry_run=payload.dry_run, diff --git a/app/modules/legacy_mysql/schemas.py b/app/modules/legacy_mysql/schemas.py index 67b58cb..fe3e397 100644 --- a/app/modules/legacy_mysql/schemas.py +++ b/app/modules/legacy_mysql/schemas.py @@ -3,12 +3,20 @@ from typing import Any from pydantic import BaseModel, Field from app.core.constants import ActorValue +from app.modules.legacy_mysql.constants import LegacyQueryName class ReadonlyQueryRequest(BaseModel): - """Readonly SQL query request for the legacy MySQL connection.""" + """Allowlisted readonly query request for the legacy MySQL connection.""" - sql: str = Field(..., description="Readonly SELECT statement.") + query_name: str | None = Field( + default=LegacyQueryName.PROJECTS, + description="Configured readonly query name.", + ) + sql: str | None = Field( + default=None, + description="Deprecated: must exactly match a configured readonly query.", + ) params: dict[str, Any] = Field(default_factory=dict) limit: int = Field(default=100, ge=1, le=500) @@ -32,7 +40,11 @@ class LegacyProjectSyncRequest(BaseModel): source_query: str | None = Field( default=None, - description="Optional SELECT query for project sync.", + description="Deprecated: must exactly match a configured readonly query.", + ) + source_query_name: str | None = Field( + default=LegacyQueryName.PROJECTS, + description="Configured readonly query name for project sync.", ) field_map: dict[str, str] = Field( default_factory=dict, diff --git a/app/modules/legacy_mysql/service.py b/app/modules/legacy_mysql/service.py index 38e58c7..597d90a 100644 --- a/app/modules/legacy_mysql/service.py +++ b/app/modules/legacy_mysql/service.py @@ -19,6 +19,7 @@ from app.modules.audit.service import AuditService from app.modules.business.constants import BusinessDomain, SourceSystem, StatusValue from app.modules.business.models import LegacySyncRun, Project from app.modules.business.service import serialize_model +from app.modules.legacy_mysql.constants import LegacyQueryError, LegacyQueryName FORBIDDEN_SQL_TOKENS = { "insert", @@ -50,6 +51,18 @@ def _row_to_dict(row: RowMapping) -> dict[str, Any]: return {key: _jsonable(value) for key, value in row.items()} +def _normalize_sql(sql: str) -> str: + return " ".join(sql.strip().rstrip(";").split()).lower() + + +def _query_name_text(query_name: str | LegacyQueryName | None) -> str: + if query_name is None: + return LegacyQueryName.PROJECTS.value + if isinstance(query_name, LegacyQueryName): + return query_name.value + return str(query_name) + + class LegacyMySQLService: """Read legacy MySQL data and sync projects into the internal ledger.""" @@ -74,6 +87,17 @@ class LegacyMySQLService: if tokens & FORBIDDEN_SQL_TOKENS: raise HTTPException(status_code=400, detail="Forbidden SQL token in readonly query") + @staticmethod + def _allowed_queries() -> dict[str, str]: + settings = get_settings() + queries = { + _query_name_text(name): sql + for name, sql in settings.legacy_allowed_queries.items() + } + if settings.legacy_project_query: + queries.setdefault(LegacyQueryName.PROJECTS.value, settings.legacy_project_query) + return queries + def health(self) -> dict[str, str]: engine = self._ensure_engine() try: @@ -113,6 +137,39 @@ class LegacyMySQLService: sql: str, params: dict[str, Any] | None = None, limit: int = 100, + ) -> dict[str, Any]: + """Execute a readonly SQL statement only when it matches the allowlist.""" + + normalized_sql = _normalize_sql(sql) + for allowed_sql in self._allowed_queries().values(): + if _normalize_sql(allowed_sql) == normalized_sql: + return self._execute_readonly_sql(allowed_sql, params, limit) + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=LegacyQueryError.QUERY_NOT_ALLOWED, + ) + + def execute_allowed_query( + self, + query_name: str | None, + params: dict[str, Any] | None = None, + limit: int = 100, + ) -> dict[str, Any]: + queries = self._allowed_queries() + normalized_name = _query_name_text(query_name) + sql = queries.get(normalized_name) + if not sql: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=LegacyQueryError.QUERY_NOT_ALLOWED, + ) + return self._execute_readonly_sql(sql, params, limit) + + def _execute_readonly_sql( + self, + sql: str, + params: dict[str, Any] | None = None, + limit: int = 100, ) -> dict[str, Any]: self._ensure_readonly(sql) engine = self._ensure_engine() @@ -132,9 +189,9 @@ class LegacyMySQLService: if not settings.legacy_project_query: raise HTTPException( status_code=400, - detail="LEGACY_PROJECT_QUERY is not configured. Configure it in .env first.", + detail=LegacyQueryError.PROJECT_QUERY_NOT_CONFIGURED, ) - return self.execute_readonly(settings.legacy_project_query, {"limit": limit}, limit=limit) + return self.execute_allowed_query(LegacyQueryName.PROJECTS, {"limit": limit}, limit=limit) @staticmethod def _value( @@ -187,6 +244,7 @@ class LegacyMySQLService: def sync_projects( self, source_query: str | None = None, + source_query_name: str | None = None, field_map: dict[str, str] | None = None, limit: int = 100, dry_run: bool = True, @@ -198,12 +256,13 @@ class LegacyMySQLService: detail="Application database session is not available", ) - settings = get_settings() - query = source_query or settings.legacy_project_query - if not query: - raise HTTPException(status_code=400, detail="Project sync query is not configured") - - rows = self.execute_readonly(query, {"limit": limit}, limit=limit)["rows"] + query_name = source_query_name or LegacyQueryName.PROJECTS + if source_query: + rows = self.execute_readonly(source_query, {"limit": limit}, limit=limit)["rows"] + query_ref = "allowlisted_inline_sql" + else: + rows = self.execute_allowed_query(query_name, {"limit": limit}, limit=limit)["rows"] + query_ref = _query_name_text(query_name) field_map = field_map or {} created = 0 updated = 0 @@ -288,7 +347,7 @@ class LegacyMySQLService: target_type=BusinessDomain.PROJECTS, risk_level=AuditRiskLevel.MEDIUM, request_payload={ - "source_query": source_query or "LEGACY_PROJECT_QUERY", + "source_query": query_ref, "field_map": field_map, "limit": limit, "dry_run": dry_run, diff --git a/app/modules/reports/service.py b/app/modules/reports/service.py index c5686f7..22e915f 100644 --- a/app/modules/reports/service.py +++ b/app/modules/reports/service.py @@ -900,19 +900,13 @@ class ReportService: WorkTask.due_date >= start, WorkTask.due_date <= end, ] - start_at = datetime.combine(start, datetime.min.time()) - end_at = datetime.combine(end, datetime.max.time()) project_filters = [] risk_filters = [RiskEvent.status == StatusValue.OPEN] procurement_filters = [ Procurement.approval_status.in_(PENDING_APPROVAL_STATUSES), - Procurement.created_at >= start_at, - Procurement.created_at <= end_at, ] expense_filters = [ Expense.approval_status.in_(PENDING_APPROVAL_STATUSES), - Expense.created_at >= start_at, - Expense.created_at <= end_at, ] attendance_filters = [ AttendanceRecord.work_date >= start, diff --git a/logs/feishu_long_connection.err.log b/logs/feishu_long_connection.err.log deleted file mode 100644 index a63abdc..0000000 --- a/logs/feishu_long_connection.err.log +++ /dev/null @@ -1,14 +0,0 @@ -INFO:__main__:Starting Feishu long connection client -INFO:httpx:HTTP Request: POST https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal "HTTP/1.1 200 OK" -INFO:httpx:HTTP Request: POST https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=chat_id "HTTP/1.1 200 OK" -INFO:__main__:Handled Feishu long connection event: {'ok': True, 'handled': True, 'result': {'command': 'risk_summary', 'reply_type': 'card', 'title': '风险预警', 'content': '- 综合风险等级：low\n- 风险分：0\n- 逾期任务：0\n- 延期项目：0\n- 超预算项目：0\n- 资金风险账户：0', 'lines': ['- 综合风险等级：low', '- 风险分：0', '- 逾期任务：0', '- 延期项目：0', '- 超预算项目：0', '- 资金风险账户：0'], 'provider_response': {'code': 0, 'data': {'body': {'content': '{"title":"风险预警","elements":[[{"tag":"text","text":"- 综合风险等级：low\\n- 风险分：0\\n- 逾期任务：0\\n- 延期项目：0\\n- 超预算项目：0\\n- 资金风险账户：0"}]]}'}, 'chat_id': 'oc_4ea0ed9fbfe4da2aca67cca9ee22cb36', 'create_time': '1782103535265', 'deleted': False, 'message_id': 'om_x100b6cbf902318acb15878ba4b86f90', 'msg_type': 'interactive', 'sender': {'id': 'cli_aab22b1674799bef', 'id_type': 'app_id', 'sender_type': 'app', 'tenant_key': '1b3485e1c8ea174f'}, 'update_time': '1782103535265', 'updated': False}, 'msg': 'success'}}} -INFO:httpx:HTTP Request: POST https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal "HTTP/1.1 200 OK" -INFO:httpx:HTTP Request: POST https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=chat_id "HTTP/1.1 200 OK" -INFO:__main__:Handled Feishu long connection event: {'ok': True, 'handled': True, 'result': {'command': 'risk_summary', 'reply_type': 'card', 'title': '风险预警', 'content': '- 综合风险等级：low\n- 风险分：0\n- 逾期任务：0\n- 延期项目：0\n- 超预算项目：0\n- 资金风险账户：0', 'lines': ['- 综合风险等级：low', '- 风险分：0', '- 逾期任务：0', '- 延期项目：0', '- 超预算项目：0', '- 资金风险账户：0'], 'provider_response': {'code': 0, 'data': {'body': {'content': '{"title":"风险预警","elements":[[{"tag":"text","text":"- 综合风险等级：low\\n- 风险分：0\\n- 逾期任务：0\\n- 延期项目：0\\n- 超预算项目：0\\n- 资金风险账户：0"}]]}'}, 'chat_id': 'oc_4ea0ed9fbfe4da2aca67cca9ee22cb36', 'create_time': '1782103549102', 'deleted': False, 'message_id': 'om_x100b6cbf91060cacb1fa837ac0932ee', 'msg_type': 'interactive', 'sender': {'id': 'cli_aab22b1674799bef', 'id_type': 'app_id', 'sender_type': 'app', 'tenant_key': '1b3485e1c8ea174f'}, 'update_time': '1782103549102', 'updated': False}, 'msg': 'success'}}} -INFO:httpx:HTTP Request: POST https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal "HTTP/1.1 200 OK" -INFO:httpx:HTTP Request: POST https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=chat_id "HTTP/1.1 200 OK" -INFO:__main__:Handled Feishu long connection event: {'ok': True, 'handled': True, 'result': {'command': 'fallback_ai', 'reply_type': 'text', 'title': 'AI 回复', 'content': 'AI provider is not configured yet. This is a deterministic placeholder. Set MODEL_PROVIDER to openclaw, hermes, or direct_llm after credentials are ready.', 'provider_response': {'code': 0, 'data': {'body': {'content': '{"text":"AI provider is not configured yet. This is a deterministic placeholder. Set MODEL_PROVIDER to openclaw, hermes, or direct_llm after credentials are ready."}'}, 'chat_id': 'oc_4ea0ed9fbfe4da2aca67cca9ee22cb36', 'create_time': '1782103556285', 'deleted': False, 'message_id': 'om_x100b6cbfae8fc8acb03feca923be229', 'msg_type': 'text', 'sender': {'id': 'cli_aab22b1674799bef', 'id_type': 'app_id', 'sender_type': 'app', 'tenant_key': '1b3485e1c8ea174f'}, 'update_time': '1782103556285', 'updated': False}, 'msg': 'success'}}} -INFO:httpx:HTTP Request: POST https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal "HTTP/1.1 200 OK" -INFO:httpx:HTTP Request: POST https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=chat_id "HTTP/1.1 200 OK" -INFO:__main__:Handled Feishu long connection event: {'ok': True, 'handled': True, 'result': {'command': 'risk_summary', 'reply_type': 'card', 'title': '风险预警', 'content': '- 综合风险等级：low\n- 风险分：0\n- 逾期任务：0\n- 延期项目：0\n- 超预算项目：0\n- 资金风险账户：0', 'lines': ['- 综合风险等级：low', '- 风险分：0', '- 逾期任务：0', '- 延期项目：0', '- 超预算项目：0', '- 资金风险账户：0'], 'provider_response': {'code': 0, 'data': {'body': {'content': '{"title":"风险预警","elements":[[{"tag":"text","text":"- 综合风险等级：low\\n- 风险分：0\\n- 逾期任务：0\\n- 延期项目：0\\n- 超预算项目：0\\n- 资金风险账户：0"}]]}'}, 'chat_id': 'oc_4ea0ed9fbfe4da2aca67cca9ee22cb36', 'create_time': '1782105193432', 'deleted': False, 'message_id': 'om_x100b6cb8085c94a4b12d99465abb1ca', 'msg_type': 'interactive', 'sender': {'id': 'cli_aab22b1674799bef', 'id_type': 'app_id', 'sender_type': 'app', 'tenant_key': '1b3485e1c8ea174f'}, 'update_time': '1782105193432', 'updated': False}, 'msg': 'success'}}} -ERROR:Lark:receive message loop exit, err: no close frame received or sent [conn_id=7654075862284225524] diff --git a/logs/feishu_long_connection.out.log b/logs/feishu_long_connection.out.log deleted file mode 100644 index 2ab17a0..0000000 --- a/logs/feishu_long_connection.out.log +++ /dev/null @@ -1 +0,0 @@ -[Lark] [2026-06-22 13:36:22,942] [ERROR] receive message loop exit, err: no close frame received or sent [conn_id=7654075862284225524] diff --git a/tests/test_ai_adapters.py b/tests/test_ai_adapters.py index f2c7495..ea180e4 100644 --- a/tests/test_ai_adapters.py +++ b/tests/test_ai_adapters.py @@ -208,3 +208,32 @@ def test_openclaw_adapter_blocks_tools_not_in_allowlist(monkeypatch) -> None: assert exc_info.value.status_code == 403 assert DummyClient.calls == [] + + +def test_openclaw_hermes_adapter_fails_when_requested_tool_is_blocked(monkeypatch) -> None: + DummyClient.calls = [] + monkeypatch.setattr(adapters.httpx, "Client", DummyClient) + settings = Settings( + model_provider="openclaw_hermes", + openclaw_http_url="http://openclaw.local", + openclaw_gateway_token="openclaw-key", + openclaw_allowed_tools=["sessions_list"], + hermes_base_url="http://hermes.local/v1", + hermes_api_key="hermes-key", + ) + + with pytest.raises(HTTPException) as exc_info: + adapters.OpenClawHermesAdapter(settings).ask( + "write through gateway", + { + AIContextKey.OPENCLAW_TOOL: "filesystem_write", + AIContextKey.OPENCLAW_ARGS: {"path": "/tmp/x"}, + }, + ) + + assert exc_info.value.status_code == 403 + assert [call["url"] for call in DummyClient.calls] == [ + "http://hermes.local/v1/chat/completions", + "http://openclaw.local/healthz", + "http://openclaw.local/readyz", + ] diff --git a/tests/test_smoke.py b/tests/test_smoke.py index 3f98fa3..22c655d 100644 --- a/tests/test_smoke.py +++ b/tests/test_smoke.py @@ -15,6 +15,8 @@ _db.close() os.environ["DATABASE_URL"] = "sqlite:///" + _db.name.replace("\\", "/") os.environ["API_KEY"] = "test-key" +os.environ["APPROVAL_API_KEY"] = "approval-key" +os.environ["APPROVAL_API_ACTOR"] = "approval-manager" os.environ["FEISHU_APP_ID"] = "" os.environ["FEISHU_APP_SECRET"] = "" os.environ["FEISHU_VERIFICATION_TOKEN"] = "test-feishu-token" @@ -25,7 +27,7 @@ from fastapi.testclient import TestClient from app.core.config import get_settings from app.core.database import Base, engine -from app.core.security import require_api_key +from app.core.security import require_api_key, require_approval_api_key from app.main import app from app.modules.legacy_mysql.service import LegacyMySQLService from app.modules.reports.constants import ( @@ -41,6 +43,7 @@ from app.modules.reports.constants import ( Base.metadata.create_all(bind=engine) client = TestClient(app) headers = {"X-API-Key": "test-key"} +approval_headers = {"X-API-Key": "test-key", "X-Approval-API-Key": "approval-key"} def teardown_module() -> None: @@ -118,12 +121,25 @@ def test_api_key_and_feishu_webhook_fail_closed(monkeypatch) -> None: json={"schema": "2.0", "header": {"event_type": "im.message.receive_v1"}}, ) assert response.status_code == 401 + + monkeypatch.setenv("APPROVAL_API_KEY", "") + get_settings.cache_clear() + with pytest.raises(HTTPException) as approval_exc_info: + require_approval_api_key("approval-key") + assert approval_exc_info.value.status_code == 503 finally: monkeypatch.setenv("API_KEY", "test-key") + monkeypatch.setenv("APPROVAL_API_KEY", "approval-key") get_settings.cache_clear() def test_approval_gate_for_high_risk_update() -> None: + create_payload = { + "code": "FUND-SMOKE-001", + "name": "Main Account", + "current_balance": 1000, + "safety_line": 500, + } blocked_create_response = client.post( "/api/v1/business/fund-accounts", headers=headers, @@ -145,7 +161,7 @@ def test_approval_gate_for_high_risk_update() -> None: "action": "create:fund-accounts", "applicant": "spoofed-user", "reason": "Smoke test account creation", - "payload": {"code": "FUND-SMOKE-001"}, + "payload": create_payload, }, ) assert create_approval_response.status_code == 200 @@ -154,11 +170,11 @@ def test_approval_gate_for_high_risk_update() -> None: approve_create_response = client.post( f"/api/v1/approvals/{create_ticket_id}/approve", - headers=headers, + headers=approval_headers, json={"approver": "spoofed-manager", "comment": "ok"}, ) assert approve_create_response.status_code == 200 - assert approve_create_response.json()["approver"] == "api" + assert approve_create_response.json()["approver"] == "approval-manager" create_response = client.post( "/api/v1/business/fund-accounts", @@ -166,17 +182,25 @@ def test_approval_gate_for_high_risk_update() -> None: json={ "actor": "spoofed-user", "approval_ticket_id": create_ticket_id, - "data": { - "code": "FUND-SMOKE-001", - "name": "Main Account", - "current_balance": 1000, - "safety_line": 500, - }, + "data": create_payload, }, ) assert create_response.status_code == 200 record_id = create_response.json()["data"]["id"] + reuse_create_response = client.post( + "/api/v1/business/fund-accounts", + headers=headers, + json={ + "approval_ticket_id": create_ticket_id, + "data": { + "code": "FUND-SMOKE-REUSE", + "name": "Reuse Account", + }, + }, + ) + assert reuse_create_response.status_code == 403 + blocked_response = client.patch( f"/api/v1/business/fund-accounts/{record_id}", headers=headers, @@ -212,12 +236,23 @@ def test_approval_gate_for_high_risk_update() -> None: approve_response = client.post( f"/api/v1/approvals/{ticket_id}/approve", - headers=headers, + headers=approval_headers, json={"approver": "spoofed-manager", "comment": "ok"}, ) assert approve_response.status_code == 200 assert approve_response.json()["status"] == "approved" - assert approve_response.json()["approver"] == "api" + assert approve_response.json()["approver"] == "approval-manager" + + mismatch_response = client.patch( + f"/api/v1/business/fund-accounts/{record_id}", + headers=headers, + json={ + "actor": "spoofed-user", + "approval_ticket_id": ticket_id, + "data": {"current_balance": 101}, + }, + ) + assert mismatch_response.status_code == 403 update_response = client.patch( f"/api/v1/business/fund-accounts/{record_id}", @@ -231,6 +266,17 @@ def test_approval_gate_for_high_risk_update() -> None: assert update_response.status_code == 200 assert update_response.json()["data"]["current_balance"] == 100.0 + reuse_update_response = client.patch( + f"/api/v1/business/fund-accounts/{record_id}", + headers=headers, + json={ + "actor": "spoofed-user", + "approval_ticket_id": ticket_id, + "data": {"current_balance": 100}, + }, + ) + assert reuse_update_response.status_code == 403 + def test_new_ledgers_reports_and_risk_events() -> None: domains_response = client.get("/api/v1/business/domains", headers=headers) @@ -437,6 +483,59 @@ def test_project_lifecycle_report_summarizes_progress_cost_and_risk() -> None: ) +def test_work_report_counts_pending_approval_backlog_outside_period() -> None: + today = date.today() + project_code = "P-BACKLOG-001" + old_created_at = (today - timedelta(days=30)).isoformat() + "T00:00:00" + + procurement_response = client.post( + "/api/v1/business/procurements", + headers=headers, + json={ + "data": { + "code": "PROC-BACKLOG-001", + "name": "Backlog procurement", + "project_code": project_code, + "approval_status": StatusValue.PENDING_APPROVAL, + "created_at": old_created_at, + }, + }, + ) + assert procurement_response.status_code == 200 + + expense_response = client.post( + "/api/v1/business/expenses", + headers=headers, + json={ + "data": { + "code": "EXP-BACKLOG-001", + "expense_type": "鍔炲叕", + "amount": 50, + "project_code": project_code, + "approval_status": StatusValue.PENDING_APPROVAL, + "created_at": old_created_at, + }, + }, + ) + assert expense_response.status_code == 200 + + report_response = client.post( + "/api/v1/reports/work-reports/generate", + headers=headers, + json={ + "report_type": ReportType.DAILY, + "project_code": project_code, + "period_start": today.isoformat(), + "period_end": today.isoformat(), + "persist": False, + }, + ) + assert report_response.status_code == 200 + metrics = report_response.json()["report"]["metrics"] + assert metrics["procurements_pending"] == 1 + assert metrics["expenses_pending"] == 1 + + def test_ai_noop_provider() -> None: response = client.post( "/api/v1/ai/ask", @@ -456,3 +555,30 @@ def test_legacy_project_payload_does_not_create_legacy_none_code() -> None: assert payload["code"] is None assert payload["external_id"] is None + + +def test_legacy_readonly_query_requires_allowlist(monkeypatch) -> None: + monkeypatch.delenv("LEGACY_PROJECT_QUERY", raising=False) + get_settings.cache_clear() + try: + with pytest.raises(HTTPException) as blocked_exc_info: + LegacyMySQLService(None).execute_readonly("SELECT id FROM secret_projects") + assert blocked_exc_info.value.status_code == 403 + + monkeypatch.setenv("LEGACY_PROJECT_QUERY", "SELECT id FROM projects") + get_settings.cache_clear() + + def unavailable_engine(): + raise HTTPException(status_code=503, detail="legacy unavailable") + + monkeypatch.setattr( + LegacyMySQLService, + "_ensure_engine", + staticmethod(unavailable_engine), + ) + with pytest.raises(HTTPException) as engine_exc_info: + LegacyMySQLService(None).execute_readonly("SELECT id FROM projects") + assert engine_exc_info.value.status_code == 503 + finally: + monkeypatch.delenv("LEGACY_PROJECT_QUERY", raising=False) + get_settings.cache_clear()