# SSP 控制項 / 檢查項目（AO）現況說明功能 — Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** 讓 PM 在 GRC 專案中填寫 SSP 控制項層級和 AO（檢查項目）層級的現況實作描述，並可附加關聯程序書文件。 **Architecture:** 擴展 jedi-oscal 套件的既有 `ControlImplementation` DDD 棧（Model → Mapper → Entity → Repo → DomainService），新增 `control_identifier`、`control_origination`、`remarks` 欄位和 `SspControlImplementationObjective` 全棧。主專案新增 SSP Route + App Service + Serializer。專案啟動流程加入 Step 7.5 初始化邏輯。 **Tech Stack:** Python 3.11, Flask-RESTful, SQLAlchemy 2.0 (mapped_column), dependency-injector, marshmallow, PostgreSQL (JSONB) **修改範圍總覽：** | 區域 | 套件 | 新增/修改 | 檔案數 | |------|------|----------|--------| | ORM Model | jedi-oscal | 修改 1 + 新增 1 | 2 | | Entity | jedi-oscal | 修改 2 + 新增 2 | 4 | | Mapper | jedi-oscal | 修改 1 + 新增 1 | 2 | | Repo | jedi-oscal | 修改 1 + 新增 2 | 3 | | Domain Service | jedi-oscal | 修改 1 + 新增 1 | 2 | | App DTO | jedi-oscal | 修改 1 + 新增 1 | 2 | | __init__ exports | jedi-oscal | 修改 5 | 5 | | Error Code | 主專案 | 修改 1 | 1 | | Serializer | 主專案 | 修改 1 + 新增 1 | 2 | | App Service | 主專案 | 新增 1 | 1 | | Route | 主專案 | 新增 1 | 1 | | DI Container | 主專案 | 修改 1 | 1 | | Blueprint 註冊 | 主專案 | 修改 1 | 1 | | 專案啟動流程 | 主專案 | 修改 1 | 1 | | Migration SQL | 主專案 | 新增 1 | 1 | | **合計** | | | **~30** | --- ## File Structure ### jedi-oscal 套件 (`~/Projects/Jedicogy/module/jedi-python-package/jedi-oscal/jedi_oscal/`) **修改：** - `infra/model/ssp/ssp_control_implementation.py` — 加欄位 `control_identifier`, `control_origination`, `remarks`；`catalog_control_id` 改 nullable；新增 `objectives` relationship - `infra/mapper/ssp/control_implementation_mapper.py` — 新增欄位映射 - `domain/entity/ssp/ssp_control_implementation_entity.py` — 新增欄位 - `domain/entity/ssp/ssp_control_implementation_query_entity.py` — 新增 `control_identifier` 查詢欄位 - `app/dto/ssp/ssp_control_implementation_dto.py` — 新增欄位 + objectives list - `domain/services/ssp/control_implementation_domain_service.py` — 新增 `get_by_ssp_and_identifier` 方法 - `infra/repository/ssp/ssp_control_implementation_repo_impl.py` — 新增自訂查詢方法（by ssp_id + control_identifier） **新增：** - `infra/model/ssp/ssp_control_implementation_objective.py` — ORM Model - `infra/mapper/ssp/control_implementation_objective_mapper.py` — Mapper - `domain/entity/ssp/ssp_control_impl_objective_entity.py` — Entity - `domain/entity/ssp/ssp_control_impl_objective_query_entity.py` — Query Entity - `domain/repository/ssp/control_implementation_objective_repo.py` — IRepo - `infra/repository/ssp/ssp_control_impl_objective_repo_impl.py` — Repo Impl - `domain/services/ssp/control_implementation_objective_domain_service.py` — Domain Service - `app/dto/ssp/ssp_control_impl_objective_dto.py` — DTO **__init__ exports 修改：** - `infra/model/ssp/__init__.py` - `infra/repository/ssp/__init__.py` - `domain/repository/ssp/__init__.py` - `domain/services/ssp/__init__.py` - `domain/entity/ssp/__init__.py`（如有） ### 主專案 (`compliance-manager-be/`) **修改：** - `common/code/grc_error_code.py` — 新增 SSP 相關 error code - `api/oscal/serializers/ssp/ssp_control_implementation.py` — 擴充欄位 + request schema - `api/oscal/__init__.py` — 註冊 SSP routes - `di_containers/oscal/oscal_containers.py` — 註冊 objective repo/domain service/app service - `app/project/service/oscal_project_service.py` — Step 7.5 初始化 **新增：** - `api/oscal/serializers/ssp/ssp_control_impl_objective.py` — Objective serializer - `app/oscal/service/ssp_control_implementation_service.py` — App Service - `api/oscal/routes/ssp/ssp_control_implementation_route.py` — Route - `scripts/sql/ssp_control_implementation_migration.sql` — Migration SQL --- ## Task 1: Migration SQL 與 ORM Model 變更（jedi-oscal） **Files:** - Create: `compliance-manager-be/scripts/sql/ssp_control_implementation_migration.sql` - Modify: `~/Projects/Jedicogy/module/jedi-python-package/jedi-oscal/jedi_oscal/infra/model/ssp/ssp_control_implementation.py` - Create: `~/Projects/Jedicogy/module/jedi-python-package/jedi-oscal/jedi_oscal/infra/model/ssp/ssp_control_implementation_objective.py` ### 1.1 Migration SQL - [ ] **Step 1: 建立 migration SQL** ```sql -- SSP Control Implementation 欄位異動 -- 1) catalog_control_id 改為 nullable ALTER TABLE oscal.system_security_plan_control_implementations ALTER COLUMN catalog_control_id DROP NOT NULL; -- 2) 新增欄位 ALTER TABLE oscal.system_security_plan_control_implementations ADD COLUMN IF NOT EXISTS control_identifier VARCHAR(100), ADD COLUMN IF NOT EXISTS control_origination VARCHAR(50) DEFAULT 'organization', ADD COLUMN IF NOT EXISTS remarks TEXT; -- 3) 補回填 control_identifier（從 catalog_controls 反查） UPDATE oscal.system_security_plan_control_implementations ci SET control_identifier = cc.control_id FROM oscal.catalog_controls cc WHERE ci.catalog_control_id = cc.id AND ci.control_identifier IS NULL; -- 4) control_identifier 設為 NOT NULL（回填完成後） ALTER TABLE oscal.system_security_plan_control_implementations ALTER COLUMN control_identifier SET NOT NULL; -- 5) 新增 index CREATE INDEX IF NOT EXISTS ix_ssp_ctrl_impl_control_identifier ON oscal.system_security_plan_control_implementations (system_security_plan_id, control_identifier); -- 6) 更新 unique constraint（改用 control_identifier 取代 catalog_control_id） ALTER TABLE oscal.system_security_plan_control_implementations DROP CONSTRAINT IF EXISTS uq_ssp_ctrl_impl_ssp_catalog_control; ALTER TABLE oscal.system_security_plan_control_implementations ADD CONSTRAINT uq_ssp_ctrl_impl_ssp_control_identifier UNIQUE (system_security_plan_id, control_identifier); -- ============================================================ -- SSP Control Implementation Objectives 新表 -- ============================================================ CREATE TABLE IF NOT EXISTS oscal.ssp_control_implementation_objectives ( id SERIAL PRIMARY KEY, uid UUID NOT NULL DEFAULT gen_random_uuid(), system_security_plan_id INTEGER NOT NULL REFERENCES oscal.system_security_plans(id) ON DELETE CASCADE, control_implementation_id INTEGER NOT NULL REFERENCES oscal.system_security_plan_control_implementations(id) ON DELETE CASCADE, control_identifier VARCHAR(100) NOT NULL, statement_identifier VARCHAR(100) NOT NULL, implementation_status VARCHAR(50), implementation_description TEXT, remarks TEXT, reference_documents JSONB DEFAULT '[]'::jsonb, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), created_user VARCHAR(100), updated_user VARCHAR(100), CONSTRAINT uq_ssp_obj_ctrl_impl_stmt UNIQUE (control_implementation_id, statement_identifier) ); CREATE INDEX IF NOT EXISTS ix_ssp_obj_uid ON oscal.ssp_control_implementation_objectives (uid); CREATE INDEX IF NOT EXISTS ix_ssp_obj_ctrl_impl_id ON oscal.ssp_control_implementation_objectives (control_implementation_id); CREATE INDEX IF NOT EXISTS ix_ssp_obj_ssp_ctrl_id ON oscal.ssp_control_implementation_objectives (system_security_plan_id, control_identifier); ``` 寫入 `scripts/sql/ssp_control_implementation_migration.sql`。 ### 1.2 修改 ORM Model：ControlImplementation - [ ] **Step 2: 修改 `ssp_control_implementation.py`** 在 `OscalSystemSecurityPlanControlImplementation` 類別中： 1. `catalog_control_id` 的 `nullable` 改為 `True` 2. 移除舊 `UniqueConstraint("system_security_plan_id", "catalog_control_id", ...)` 3. 新增 `UniqueConstraint("system_security_plan_id", "control_identifier", name="uq_ssp_ctrl_impl_ssp_control_identifier")` 4. 新增欄位： ```python control_identifier: Mapped[str] = mapped_column( String(100), nullable=False, comment="控制項編號，如 AC.L1-3.1.1", ) control_origination: Mapped[Optional[str]] = mapped_column( String(50), default="organization", comment="organization | system-specific | customer-configured | inherited | shared", ) remarks: Mapped[Optional[str]] = mapped_column( Text, comment="補充備註", ) ``` 5. 新增 relationship： ```python objectives: Mapped[list["OscalSspControlImplementationObjective"]] = relationship( "OscalSspControlImplementationObjective", back_populates="control_implementation", cascade="all, delete-orphan", lazy="select", ) ``` 6. 新增 Index：`Index("ix_ssp_ctrl_impl_control_identifier", "system_security_plan_id", "control_identifier")` 7. 新增 TYPE_CHECKING import for `OscalSspControlImplementationObjective` ### 1.3 新增 ORM Model：ControlImplementationObjective - [ ] **Step 3: 建立 `ssp_control_implementation_objective.py`** ```python import uuid from typing import Optional, TYPE_CHECKING from datetime import datetime from sqlalchemy import ( String, Text, UUID, ForeignKey, DateTime, func, Index, UniqueConstraint, Integer, ) from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.orm import Mapped, mapped_column, relationship from jedi_common.session.database.declarative_base import Base if TYPE_CHECKING: from jedi_oscal.infra.model.ssp.ssp import OscalSystemSecurityPlan from jedi_oscal.infra.model.ssp.ssp_control_implementation import ( OscalSystemSecurityPlanControlImplementation, ) class OscalSspControlImplementationObjective(Base): """SSP Control Implementation Objective (AO 層級)""" __tablename__ = "ssp_control_implementation_objectives" __table_args__ = ( UniqueConstraint( "control_implementation_id", "statement_identifier", name="uq_ssp_obj_ctrl_impl_stmt", ), Index("ix_ssp_obj_uid", "uid"), Index("ix_ssp_obj_ctrl_impl_id", "control_implementation_id"), Index("ix_ssp_obj_ssp_ctrl_id", "system_security_plan_id", "control_identifier"), {"schema": "oscal"}, ) id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True) uid: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), default=uuid.uuid4, unique=True, nullable=False, ) system_security_plan_id: Mapped[int] = mapped_column( ForeignKey("oscal.system_security_plans.id", ondelete="CASCADE"), nullable=False, ) control_implementation_id: Mapped[int] = mapped_column( ForeignKey("oscal.system_security_plan_control_implementations.id", ondelete="CASCADE"), nullable=False, ) control_identifier: Mapped[str] = mapped_column(String(100), nullable=False) statement_identifier: Mapped[str] = mapped_column(String(100), nullable=False) implementation_status: Mapped[Optional[str]] = mapped_column(String(50)) implementation_description: Mapped[Optional[str]] = mapped_column(Text) remarks: Mapped[Optional[str]] = mapped_column(Text) reference_documents: Mapped[Optional[list]] = mapped_column( JSONB, default=list, server_default="'[]'::jsonb", ) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), default=func.now(), nullable=False, ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), default=func.now(), onupdate=func.now(), nullable=False, ) created_user: Mapped[Optional[str]] = mapped_column(String(100)) updated_user: Mapped[Optional[str]] = mapped_column(String(100)) # Relationships control_implementation: Mapped["OscalSystemSecurityPlanControlImplementation"] = relationship( "OscalSystemSecurityPlanControlImplementation", back_populates="objectives", ) system_security_plan: Mapped["OscalSystemSecurityPlan"] = relationship( "OscalSystemSecurityPlan", ) ``` ### 1.4 更新 __init__ exports - [ ] **Step 4: 更新 `infra/model/ssp/__init__.py`** ```python from .ssp_control_implementation_objective import OscalSspControlImplementationObjective ``` - [ ] **Step 5: 驗證 import 無誤** Run: `cd ~/Projects/Jedicogy/module/jedi-python-package/jedi-oscal && python -c "from jedi_oscal.infra.model.ssp.ssp_control_implementation_objective import OscalSspControlImplementationObjective; print('OK')"` --- ## Task 2: Entity + Query Entity（jedi-oscal） **Files:** - Modify: `jedi_oscal/domain/entity/ssp/ssp_control_implementation_entity.py` - Modify: `jedi_oscal/domain/entity/ssp/ssp_control_implementation_query_entity.py` - Create: `jedi_oscal/domain/entity/ssp/ssp_control_impl_objective_entity.py` - Create: `jedi_oscal/domain/entity/ssp/ssp_control_impl_objective_query_entity.py` - [ ] **Step 1: 修改 `ControlImplementationEntity`** 新增 `__init__` 參數與屬性： ```python control_identifier: Optional[str] = None, control_origination: Optional[str] = None, remarks: Optional[str] = None, objectives: Optional[list] = None, # list["ControlImplementationObjectiveEntity"] ``` 在 `__init__` body 中： ```python self.control_identifier = control_identifier self.control_origination = control_origination self.remarks = remarks self.objectives = objectives or [] ``` - [ ] **Step 2: 修改 `SspControlImplementationQueryEntity`** 新增 `control_identifier: Optional[str] = None` 參數與屬性。 - [ ] **Step 3: 建立 `ssp_control_impl_objective_entity.py`** ```python from datetime import datetime from typing import Optional import uuid class ControlImplementationObjectiveEntity: """SSP Control Implementation Objective domain entity.""" def __init__( self, id: Optional[int] = None, uid: Optional[uuid.UUID] = None, system_security_plan_id: Optional[int] = None, control_implementation_id: Optional[int] = None, control_identifier: Optional[str] = None, statement_identifier: Optional[str] = None, implementation_status: Optional[str] = None, implementation_description: Optional[str] = None, remarks: Optional[str] = None, reference_documents: Optional[list] = None, created_user: Optional[str] = None, updated_user: Optional[str] = None, created_at: Optional[datetime] = None, updated_at: Optional[datetime] = None, ): self.id = id self.uid = uid self.system_security_plan_id = system_security_plan_id self.control_implementation_id = control_implementation_id self.control_identifier = control_identifier self.statement_identifier = statement_identifier self.implementation_status = implementation_status self.implementation_description = implementation_description self.remarks = remarks self.reference_documents = reference_documents or [] self.created_user = created_user self.updated_user = updated_user self.created_at = created_at self.updated_at = updated_at ``` - [ ] **Step 4: 建立 `ssp_control_impl_objective_query_entity.py`** ```python from datetime import datetime from typing import Optional import uuid class SspControlImplObjectiveQueryEntity: """Query container for filtering ControlImplementationObjective.""" def __init__( self, id: Optional[int] = None, uid: Optional[uuid.UUID] = None, system_security_plan_id: Optional[int] = None, control_implementation_id: Optional[int] = None, control_identifier: Optional[str] = None, statement_identifier: Optional[str] = None, implementation_status: Optional[str] = None, ): self.id = id self.uid = uid self.system_security_plan_id = system_security_plan_id self.control_implementation_id = control_implementation_id self.control_identifier = control_identifier self.statement_identifier = statement_identifier self.implementation_status = implementation_status ``` --- ## Task 3: Mapper（jedi-oscal） **Files:** - Modify: `jedi_oscal/infra/mapper/ssp/control_implementation_mapper.py` - Create: `jedi_oscal/infra/mapper/ssp/control_implementation_objective_mapper.py` - [ ] **Step 1: 修改 `ControlImplementationMapper`** 在 `to_entity()` 中新增： ```python control_identifier=getattr(model, "control_identifier", None), control_origination=getattr(model, "control_origination", None), remarks=getattr(model, "remarks", None), objectives=ControlImplementationObjectiveMapper.to_list_entity( getattr(model, "objectives", []) or [] ), ``` 在 `to_model()` 中新增： ```python m.control_identifier = getattr(entity, "control_identifier", None) m.control_origination = getattr(entity, "control_origination", None) m.remarks = getattr(entity, "remarks", None) ``` 新增 import： ```python from jedi_oscal.infra.mapper.ssp.control_implementation_objective_mapper import ControlImplementationObjectiveMapper ``` - [ ] **Step 2: 建立 `control_implementation_objective_mapper.py`** ```python from typing import List, Optional from jedi_oscal.domain.entity.ssp.ssp_control_impl_objective_entity import ControlImplementationObjectiveEntity from jedi_oscal.infra.model.ssp.ssp_control_implementation_objective import OscalSspControlImplementationObjective class ControlImplementationObjectiveMapper: @staticmethod def to_entity(model: OscalSspControlImplementationObjective) -> ControlImplementationObjectiveEntity: return ControlImplementationObjectiveEntity( id=getattr(model, "id", None), uid=getattr(model, "uid", None), system_security_plan_id=getattr(model, "system_security_plan_id", None), control_implementation_id=getattr(model, "control_implementation_id", None), control_identifier=getattr(model, "control_identifier", None), statement_identifier=getattr(model, "statement_identifier", None), implementation_status=getattr(model, "implementation_status", None), implementation_description=getattr(model, "implementation_description", None), remarks=getattr(model, "remarks", None), reference_documents=getattr(model, "reference_documents", None) or [], created_user=getattr(model, "created_user", None), updated_user=getattr(model, "updated_user", None), created_at=getattr(model, "created_at", None), updated_at=getattr(model, "updated_at", None), ) @staticmethod def to_list_entity(models: list) -> list[ControlImplementationObjectiveEntity]: return [ControlImplementationObjectiveMapper.to_entity(m) for m in models] @staticmethod def to_model( entity: ControlImplementationObjectiveEntity, model: Optional[OscalSspControlImplementationObjective] = None, ) -> OscalSspControlImplementationObjective: m = model or OscalSspControlImplementationObjective() if getattr(entity, "uid", None) is not None: m.uid = entity.uid m.system_security_plan_id = entity.system_security_plan_id m.control_implementation_id = entity.control_implementation_id m.control_identifier = entity.control_identifier m.statement_identifier = entity.statement_identifier m.implementation_status = getattr(entity, "implementation_status", None) m.implementation_description = getattr(entity, "implementation_description", None) m.remarks = getattr(entity, "remarks", None) m.reference_documents = getattr(entity, "reference_documents", None) or [] if hasattr(m, "created_user"): m.created_user = getattr(entity, "created_user", None) if hasattr(m, "updated_user"): m.updated_user = getattr(entity, "updated_user", None) return m @staticmethod def to_list_model(entities: list) -> list[OscalSspControlImplementationObjective]: return [ControlImplementationObjectiveMapper.to_model(e) for e in entities] ``` --- ## Task 4: Repository Interface + Impl（jedi-oscal） **Files:** - Create: `jedi_oscal/domain/repository/ssp/control_implementation_objective_repo.py` - Create: `jedi_oscal/infra/repository/ssp/ssp_control_impl_objective_repo_impl.py` - Modify: `jedi_oscal/infra/repository/ssp/ssp_control_implementation_repo_impl.py` - Modify: `jedi_oscal/infra/repository/ssp/__init__.py` - Modify: `jedi_oscal/domain/repository/ssp/__init__.py` - [ ] **Step 1: 建立 IRepo `control_implementation_objective_repo.py`** ```python from jedi_common.session.database.repository.base_repository import IBaseRepo class IControlImplementationObjectiveRepo(IBaseRepo): pass ``` - [ ] **Step 2: 建立 Repo Impl `ssp_control_impl_objective_repo_impl.py`** ```python from jedi_common.session.database.repository.base_repository_impl import BaseRepositoryImpl from jedi_oscal.domain.entity.ssp.ssp_control_impl_objective_entity import ControlImplementationObjectiveEntity from jedi_oscal.domain.entity.ssp.ssp_control_impl_objective_query_entity import SspControlImplObjectiveQueryEntity from jedi_oscal.domain.repository.ssp.control_implementation_objective_repo import IControlImplementationObjectiveRepo from jedi_oscal.infra.mapper.ssp.control_implementation_objective_mapper import ControlImplementationObjectiveMapper from jedi_oscal.infra.model.ssp.ssp_control_implementation_objective import OscalSspControlImplementationObjective class ControlImplementationObjectiveRepoImpl( IControlImplementationObjectiveRepo[ControlImplementationObjectiveEntity, SspControlImplObjectiveQueryEntity], BaseRepositoryImpl[ ControlImplementationObjectiveEntity, SspControlImplObjectiveQueryEntity, OscalSspControlImplementationObjective, ControlImplementationObjectiveMapper, ], ): def __init__(self): super().__init__( mapper=ControlImplementationObjectiveMapper, model=OscalSspControlImplementationObjective, ) ``` - [ ] **Step 3: 修改 ControlImplementation Repo Impl** 在 `ControlImplementationRepoImpl` 中新增自訂方法，以支援 `ssp_id + control_identifier` 的查詢： ```python from jedi_oscal.infra.model.ssp.ssp_control_implementation import OscalSystemSecurityPlanControlImplementation def get_by_ssp_and_identifier(self, ssp_id: int, control_identifier: str): """依 ssp_id + control_identifier 取得單筆，含 objectives""" model = self.session.query(OscalSystemSecurityPlanControlImplementation).filter( OscalSystemSecurityPlanControlImplementation.system_security_plan_id == ssp_id, OscalSystemSecurityPlanControlImplementation.control_identifier == control_identifier, ).first() if not model: return None return self.mapper.to_entity(model) def get_all_by_ssp_id(self, ssp_id: int): """取得某 SSP 下所有 control implementations""" models = self.session.query(OscalSystemSecurityPlanControlImplementation).filter( OscalSystemSecurityPlanControlImplementation.system_security_plan_id == ssp_id, ).all() return self.mapper.to_list_entity(models) ``` 同時在 `IControlImplementationRepo` 介面加上這兩個 abstract method： ```python from abc import abstractmethod @abstractmethod def get_by_ssp_and_identifier(self, ssp_id: int, control_identifier: str): pass @abstractmethod def get_all_by_ssp_id(self, ssp_id: int): pass ``` - [ ] **Step 4: 更新 `__init__.py` exports** `infra/repository/ssp/__init__.py`： ```python from .ssp_control_impl_objective_repo_impl import ControlImplementationObjectiveRepoImpl ``` `domain/repository/ssp/__init__.py`： ```python from .control_implementation_objective_repo import IControlImplementationObjectiveRepo ``` --- ## Task 5: Domain Service（jedi-oscal） **Files:** - Modify: `jedi_oscal/domain/services/ssp/control_implementation_domain_service.py` - Create: `jedi_oscal/domain/services/ssp/control_implementation_objective_domain_service.py` - Modify: `jedi_oscal/domain/services/ssp/__init__.py` - [ ] **Step 1: 修改 `ControlImplementationDomainService`** 新增方法： ```python def get_by_ssp_and_identifier(self, ssp_id: int, control_identifier: str): return self.control_implementation_repo.get_by_ssp_and_identifier(ssp_id, control_identifier) def get_all_by_ssp_id(self, ssp_id: int): return self.control_implementation_repo.get_all_by_ssp_id(ssp_id) ``` - [ ] **Step 2: 建立 `ControlImplementationObjectiveDomainService`** ```python from jedi_oscal.domain.entity.ssp.ssp_control_impl_objective_entity import ControlImplementationObjectiveEntity from jedi_oscal.domain.entity.ssp.ssp_control_impl_objective_query_entity import SspControlImplObjectiveQueryEntity from jedi_oscal.domain.repository.ssp import IControlImplementationObjectiveRepo class ControlImplementationObjectiveDomainService: def __init__(self, repo: IControlImplementationObjectiveRepo): self.repo = repo def get_all(self, _filter: SspControlImplObjectiveQueryEntity = None): return self.repo.get_all_by_fields(_filter) def get_one(self, _filter: SspControlImplObjectiveQueryEntity): return self.repo.get_one_by_fields(_filter) def get_by_uid(self, uid: str): return self.repo.get_by_uid(uid) def get_by_id(self, id: int): return self.repo.get_by_id(id) def add(self, entity: ControlImplementationObjectiveEntity): return self.repo.add(entity) def update(self, entity: ControlImplementationObjectiveEntity): return self.repo.update(entity) def delete_by_id(self, id: int): return self.repo.delete_by_id(id) def delete_by_uid(self, uid: str): return self.repo.delete_by_uid(uid) ``` - [ ] **Step 3: 更新 `__init__.py`** ```python from .control_implementation_objective_domain_service import ControlImplementationObjectiveDomainService ``` --- ## Task 6: App DTO（jedi-oscal） **Files:** - Modify: `jedi_oscal/app/dto/ssp/ssp_control_implementation_dto.py` - Create: `jedi_oscal/app/dto/ssp/ssp_control_impl_objective_dto.py` - [ ] **Step 1: 建立 `ssp_control_impl_objective_dto.py`** ```python from __future__ import annotations from dataclasses import dataclass, field from typing import Optional from jedi_oscal.app.dto.base.oscal_base_dto import OscalBaseDTO from jedi_oscal.domain.entity.ssp.ssp_control_impl_objective_entity import ControlImplementationObjectiveEntity @dataclass class ControlImplementationObjectiveDTO(OscalBaseDTO): uid: Optional[str] = None system_security_plan_id: Optional[int] = None control_implementation_id: Optional[int] = None control_identifier: Optional[str] = None statement_identifier: Optional[str] = None implementation_status: Optional[str] = None implementation_description: Optional[str] = None remarks: Optional[str] = None reference_documents: list = field(default_factory=list) @staticmethod def from_entity(entity: ControlImplementationObjectiveEntity) -> "ControlImplementationObjectiveDTO": return ControlImplementationObjectiveDTO( id=getattr(entity, "id", None), uid=str(entity.uid) if getattr(entity, "uid", None) else None, system_security_plan_id=getattr(entity, "system_security_plan_id", None), control_implementation_id=getattr(entity, "control_implementation_id", None), control_identifier=getattr(entity, "control_identifier", None), statement_identifier=getattr(entity, "statement_identifier", None), implementation_status=getattr(entity, "implementation_status", None), implementation_description=getattr(entity, "implementation_description", None), remarks=getattr(entity, "remarks", None), reference_documents=getattr(entity, "reference_documents", None) or [], created_at=getattr(entity, "created_at", None), updated_at=getattr(entity, "updated_at", None), created_user=getattr(entity, "created_user", None), updated_user=getattr(entity, "updated_user", None), ) @staticmethod def from_entity_list(entities: list) -> list["ControlImplementationObjectiveDTO"]: return [ControlImplementationObjectiveDTO.from_entity(e) for e in entities] ``` - [ ] **Step 2: 修改 `ControlImplementationDTO`** 新增欄位： ```python control_identifier: Optional[str] = None control_origination: Optional[str] = None remarks: Optional[str] = None objectives: list = field(default_factory=list) # list[ControlImplementationObjectiveDTO] ``` 在 `from_entity()` 中新增： ```python from jedi_oscal.app.dto.ssp.ssp_control_impl_objective_dto import ControlImplementationObjectiveDTO control_identifier=getattr(entity, "control_identifier", None), control_origination=getattr(entity, "control_origination", None), remarks=getattr(entity, "remarks", None), objectives=ControlImplementationObjectiveDTO.from_entity_list( getattr(entity, "objectives", []) or [] ), ``` 在 `ControlImplementationDTO` 上方加 `from dataclasses import dataclass, field`（取代原有 `dataclass` import）。 --- ## Task 7: Error Code + Serializer（主專案） **Files:** - Modify: `compliance-manager-be/common/code/grc_error_code.py` - Modify: `compliance-manager-be/api/oscal/serializers/ssp/ssp_control_implementation.py` - Create: `compliance-manager-be/api/oscal/serializers/ssp/ssp_control_impl_objective.py` - [ ] **Step 1: 新增 Error Code** 在 `GrcErrorCode` 中新增： ```python # ── SSP Control Implementation ────────────────────────────────────── GRC_SSP_NOT_FOUND = ("SSP 系統安全計畫不存在", "GRC_404012") GRC_SSP_CONTROL_IMPL_NOT_FOUND = ("SSP 控制項實作記錄不存在", "GRC_404013") GRC_SSP_OBJECTIVE_NOT_FOUND = ("SSP 檢查項目記錄不存在", "GRC_404014") ``` - [ ] **Step 2: 建立 Objective Serializer `ssp_control_impl_objective.py`** ```python from marshmallow import Schema, fields class ObjectiveResponseSchema(Schema): uid = fields.Str() statement_identifier = fields.Str() implementation_status = fields.Str(allow_none=True) implementation_description = fields.Str(allow_none=True) remarks = fields.Str(allow_none=True) reference_documents = fields.List(fields.Dict(), load_default=[]) created_at = fields.DateTime() updated_at = fields.DateTime() class ObjectiveUpdateRequestSchema(Schema): implementation_status = fields.Str(load_default=None, allow_none=True) implementation_description = fields.Str(load_default=None, allow_none=True) remarks = fields.Str(load_default=None, allow_none=True) reference_documents = fields.List(fields.Dict(), load_default=None, allow_none=True) ``` - [ ] **Step 3: 修改 `ControlImplementationResponseSchema`** 新增欄位： ```python control_identifier = fields.Str() control_origination = fields.Str(allow_none=True) remarks = fields.Str(allow_none=True) from api.oscal.serializers.ssp.ssp_control_impl_objective import ObjectiveResponseSchema # ...（放在檔案頂部 import 區） objectives = fields.List(fields.Nested(ObjectiveResponseSchema), dump_default=[]) ``` 新增 Request Schema（同一檔案底部）： ```python class ControlImplementationUpdateRequestSchema(Schema): implementation_status = fields.Str(load_default=None, allow_none=True) implementation_description = fields.Str(load_default=None, allow_none=True) remarks = fields.Str(load_default=None, allow_none=True) class ControlImplementationListItemSchema(Schema): control_identifier = fields.Str() implementation_status = fields.Str(allow_none=True) implementation_description = fields.Str(allow_none=True) objective_count = fields.Int() objective_described_count = fields.Int() has_description = fields.Bool() class ControlImplementationListSummarySchema(Schema): total = fields.Int() described = fields.Int() not_described = fields.Int() completion_rate = fields.Float() class ControlImplementationListResponseSchema(Schema): items = fields.List(fields.Nested(ControlImplementationListItemSchema)) summary = fields.Nested(ControlImplementationListSummarySchema) ``` --- ## Task 8: App Service（主專案） **Files:** - Create: `compliance-manager-be/app/oscal/service/ssp_control_implementation_service.py` - [ ] **Step 1: 建立 App Service** ```python from typing import Optional from jedi_common.session.database.db import transaction from jedi_common.handler.exception import NotFound from common.code.grc_error_code import GrcErrorCode from jedi_oscal.domain.services.ssp.control_implementation_domain_service import ControlImplementationDomainService from jedi_oscal.domain.services.ssp.control_implementation_objective_domain_service import ControlImplementationObjectiveDomainService from jedi_oscal.domain.services.ssp.ssp_domain_service import SystemSecurityPlanDomainService from jedi_oscal.domain.entity.ssp.ssp_control_implementation_entity import ControlImplementationEntity from jedi_oscal.domain.entity.ssp.ssp_control_impl_objective_entity import ControlImplementationObjectiveEntity from jedi_oscal.domain.entity.ssp.ssp_control_impl_objective_query_entity import SspControlImplObjectiveQueryEntity from jedi_oscal.app.dto.ssp.ssp_control_implementation_dto import ControlImplementationDTO from jedi_oscal.app.dto.ssp.ssp_control_impl_objective_dto import ControlImplementationObjectiveDTO class SspControlImplementationService: def __init__( self, ssp_domain_service: SystemSecurityPlanDomainService, control_impl_domain_service: ControlImplementationDomainService, objective_domain_service: ControlImplementationObjectiveDomainService, ): self._ssp_ds = ssp_domain_service self._ctrl_ds = control_impl_domain_service self._obj_ds = objective_domain_service def _get_ssp_or_404(self, ssp_uid: str): ssp = self._ssp_ds.get_by_uid(ssp_uid) if not ssp: raise NotFound(GrcErrorCode.GRC_SSP_NOT_FOUND) return ssp def _get_or_create_control_impl(self, ssp_id: int, control_identifier: str, curr_user: str): """取得或自動建立 control implementation""" entity = self._ctrl_ds.get_by_ssp_and_identifier(ssp_id, control_identifier) if entity: return entity new_entity = ControlImplementationEntity( system_security_plan_id=ssp_id, control_identifier=control_identifier, implementation_status="unknown", # 不可為 None，ORM 欄位 nullable=False created_user=curr_user, updated_user=curr_user, ) return self._ctrl_ds.add(new_entity) @transaction def get_control_implementation(self, ssp_uid: str, control_identifier: str): """API 2.1: 取得控制項的 SSP 現況說明（含 objectives）""" ssp = self._get_ssp_or_404(ssp_uid) entity = self._ctrl_ds.get_by_ssp_and_identifier(ssp.id, control_identifier) if not entity: raise NotFound(GrcErrorCode.GRC_SSP_CONTROL_IMPL_NOT_FOUND) return ControlImplementationDTO.from_entity(entity) @transaction def update_control_implementation(self, ssp_uid: str, control_identifier: str, curr_user: str, **kwargs): """API 2.2: 更新控制項層級的 SSP 現況說明（upsert）""" ssp = self._get_ssp_or_404(ssp_uid) entity = self._get_or_create_control_impl(ssp.id, control_identifier, curr_user) if "implementation_status" in kwargs: entity.implementation_status = kwargs["implementation_status"] if "implementation_description" in kwargs: entity.implementation_description = kwargs["implementation_description"] if "remarks" in kwargs: entity.remarks = kwargs["remarks"] entity.updated_user = curr_user updated = self._ctrl_ds.update(entity) return ControlImplementationDTO.from_entity(updated) @transaction def update_objective(self, ssp_uid: str, control_identifier: str, statement_identifier: str, curr_user: str, **kwargs): """API 2.3: 更新 AO 層級的 SSP 現況說明（upsert）""" ssp = self._get_ssp_or_404(ssp_uid) ctrl = self._get_or_create_control_impl(ssp.id, control_identifier, curr_user) # 查詢是否已存在 query = SspControlImplObjectiveQueryEntity( control_implementation_id=ctrl.id, statement_identifier=statement_identifier, ) obj_entity = self._obj_ds.get_one(query) if obj_entity: if "implementation_status" in kwargs: obj_entity.implementation_status = kwargs["implementation_status"] if "implementation_description" in kwargs: obj_entity.implementation_description = kwargs["implementation_description"] if "remarks" in kwargs: obj_entity.remarks = kwargs["remarks"] if "reference_documents" in kwargs and kwargs["reference_documents"] is not None: obj_entity.reference_documents = kwargs["reference_documents"] obj_entity.updated_user = curr_user updated = self._obj_ds.update(obj_entity) else: new_obj = ControlImplementationObjectiveEntity( system_security_plan_id=ssp.id, control_implementation_id=ctrl.id, control_identifier=control_identifier, statement_identifier=statement_identifier, implementation_status=kwargs.get("implementation_status"), implementation_description=kwargs.get("implementation_description"), remarks=kwargs.get("remarks"), reference_documents=kwargs.get("reference_documents") or [], created_user=curr_user, updated_user=curr_user, ) updated = self._obj_ds.add(new_obj) return ControlImplementationObjectiveDTO.from_entity(updated) @transaction def add_objective_document(self, ssp_uid: str, control_identifier: str, statement_identifier: str, file_ref: dict, curr_user: str): """API 2.4: 新增程序書關聯到 objective 的 reference_documents""" ssp = self._get_ssp_or_404(ssp_uid) ctrl = self._get_or_create_control_impl(ssp.id, control_identifier, curr_user) query = SspControlImplObjectiveQueryEntity( control_implementation_id=ctrl.id, statement_identifier=statement_identifier, ) obj_entity = self._obj_ds.get_one(query) if not obj_entity: raise NotFound(GrcErrorCode.GRC_SSP_OBJECTIVE_NOT_FOUND) docs = obj_entity.reference_documents or [] docs.append(file_ref) obj_entity.reference_documents = docs obj_entity.updated_user = curr_user updated = self._obj_ds.update(obj_entity) return ControlImplementationObjectiveDTO.from_entity(updated) @transaction def remove_objective_document(self, ssp_uid: str, control_identifier: str, statement_identifier: str, file_id: str, curr_user: str): """API 2.5: 移除程序書關聯""" ssp = self._get_ssp_or_404(ssp_uid) ctrl = self._ctrl_ds.get_by_ssp_and_identifier(ssp.id, control_identifier) if not ctrl: raise NotFound(GrcErrorCode.GRC_SSP_CONTROL_IMPL_NOT_FOUND) query = SspControlImplObjectiveQueryEntity( control_implementation_id=ctrl.id, statement_identifier=statement_identifier, ) obj_entity = self._obj_ds.get_one(query) if not obj_entity: raise NotFound(GrcErrorCode.GRC_SSP_OBJECTIVE_NOT_FOUND) docs = obj_entity.reference_documents or [] obj_entity.reference_documents = [d for d in docs if d.get("file_id") != file_id] obj_entity.updated_user = curr_user self._obj_ds.update(obj_entity) @transaction def list_control_implementations(self, ssp_uid: str, control_identifiers: Optional[list] = None, status: Optional[str] = None, has_description: Optional[bool] = None): """API 2.6: 批次取得多個控制項的 SSP 現況""" ssp = self._get_ssp_or_404(ssp_uid) all_ctrls = self._ctrl_ds.get_all_by_ssp_id(ssp.id) # 篩選 if control_identifiers: all_ctrls = [c for c in all_ctrls if c.control_identifier in control_identifiers] if status: all_ctrls = [c for c in all_ctrls if c.implementation_status == status] items = [] described_count = 0 for ctrl in all_ctrls: obj_count = len(ctrl.objectives) if ctrl.objectives else 0 obj_described = sum( 1 for o in (ctrl.objectives or []) if o.implementation_description ) has_desc = bool(ctrl.implementation_description) if has_description is not None and has_desc != has_description: continue items.append({ "control_identifier": ctrl.control_identifier, "implementation_status": ctrl.implementation_status, "implementation_description": ctrl.implementation_description, "objective_count": obj_count, "objective_described_count": obj_described, "has_description": has_desc, }) if has_desc: described_count += 1 total = len(items) return { "items": items, "summary": { "total": total, "described": described_count, "not_described": total - described_count, "completion_rate": round(described_count / total, 2) if total > 0 else 0, }, } ``` --- ## Task 9: Route + DI Container + Blueprint 註冊（主專案） **Files:** - Create: `compliance-manager-be/api/oscal/routes/ssp/ssp_control_implementation_route.py` - Modify: `compliance-manager-be/di_containers/oscal/oscal_containers.py` - Modify: `compliance-manager-be/api/oscal/__init__.py` - [ ] **Step 1: 建立 Route** ```python from dependency_injector.wiring import inject, Provide from flask import request from flask_apispec import MethodResource, doc, use_kwargs, marshal_with from flask_jwt_extended import jwt_required from jedi_common.session.auth.auth_context import get_user_context from api.oscal.serializers.ssp.ssp_control_implementation import ( ControlImplementationResponseSchema, ControlImplementationUpdateRequestSchema, ControlImplementationListResponseSchema, ) from api.oscal.serializers.ssp.ssp_control_impl_objective import ( ObjectiveResponseSchema, ObjectiveUpdateRequestSchema, ) from app.oscal.service.ssp_control_implementation_service import SspControlImplementationService from common.util.response_util import return_response from di_containers.containers import Containers import logging logger = logging.getLogger(__name__) class SspControlImplementationDetailRoute(MethodResource): """GET / PUT /ssp//control-implementation/""" @doc(description="取得控制項的 SSP 現況說明", tags=["SSP"]) @marshal_with(ControlImplementationResponseSchema, apply=False) @jwt_required() @inject def get( self, ssp_uid: str, control_identifier: str, ssp_ctrl_service: SspControlImplementationService = Provide[ Containers.oscal_container.ssp_control_implementation_service ], ): result = ssp_ctrl_service.get_control_implementation(ssp_uid, control_identifier) return return_response(True, ControlImplementationResponseSchema().dump(result)) @doc(description="更新控制項層級的 SSP 現況說明", tags=["SSP"]) @use_kwargs(ControlImplementationUpdateRequestSchema, location="json", apply=False) @jwt_required() @inject def put( self, ssp_uid: str, control_identifier: str, ssp_ctrl_service: SspControlImplementationService = Provide[ Containers.oscal_container.ssp_control_implementation_service ], ): user = get_user_context() payload = request.get_json(silent=True) or {} data = ControlImplementationUpdateRequestSchema().load(payload) result = ssp_ctrl_service.update_control_implementation( ssp_uid, control_identifier, user.login_name, **data ) return return_response(True, ControlImplementationResponseSchema().dump(result)) class SspControlImplementationListRoute(MethodResource): """GET /ssp//control-implementations""" @doc(description="批次取得多個控制項的 SSP 現況", tags=["SSP"]) @marshal_with(ControlImplementationListResponseSchema, apply=False) @jwt_required() @inject def get( self, ssp_uid: str, ssp_ctrl_service: SspControlImplementationService = Provide[ Containers.oscal_container.ssp_control_implementation_service ], ): control_ids_str = request.args.get("control_identifiers") control_ids = control_ids_str.split(",") if control_ids_str else None status = request.args.get("status") has_desc_str = request.args.get("has_description") has_description = None if has_desc_str is not None: has_description = has_desc_str.lower() == "true" result = ssp_ctrl_service.list_control_implementations( ssp_uid, control_identifiers=control_ids, status=status, has_description=has_description, ) return return_response(True, ControlImplementationListResponseSchema().dump(result)) class SspObjectiveDetailRoute(MethodResource): """PUT /ssp//control-implementation//objective/""" @doc(description="更新 AO 層級的 SSP 現況說明", tags=["SSP"]) @use_kwargs(ObjectiveUpdateRequestSchema, location="json", apply=False) @jwt_required() @inject def put( self, ssp_uid: str, control_identifier: str, statement_identifier: str, ssp_ctrl_service: SspControlImplementationService = Provide[ Containers.oscal_container.ssp_control_implementation_service ], ): user = get_user_context() payload = request.get_json(silent=True) or {} data = ObjectiveUpdateRequestSchema().load(payload) result = ssp_ctrl_service.update_objective( ssp_uid, control_identifier, statement_identifier, user.login_name, **data ) return return_response(True, ObjectiveResponseSchema().dump(result)) class SspObjectiveDocumentRoute(MethodResource): """DELETE /ssp//control-implementation//objective//document/""" @doc(description="移除程序書關聯", tags=["SSP"]) @jwt_required() @inject def delete( self, ssp_uid: str, control_identifier: str, statement_identifier: str, file_id: str, ssp_ctrl_service: SspControlImplementationService = Provide[ Containers.oscal_container.ssp_control_implementation_service ], ): user = get_user_context() ssp_ctrl_service.remove_objective_document( ssp_uid, control_identifier, statement_identifier, file_id, user.login_name ) return return_response(True, {}) ``` > **Note：** 文件上傳 API（2.4）需整合 `jedi-file-upload`（ManagedFileUploadService），實作較複雜，建議在 Route 層處理 `request.files` → 上傳 → 取得 file_ref dict → 呼叫 `add_objective_document()`。如需此功能可在後續 Task 擴充。 - [ ] **Step 2: 修改 DI Container `oscal_containers.py`** 新增 imports： ```python from jedi_oscal.infra.repository.ssp import ControlImplementationRepoImpl, ControlImplementationObjectiveRepoImpl from jedi_oscal.domain.services.ssp import ControlImplementationDomainService, ControlImplementationObjectiveDomainService from app.oscal.service.ssp_control_implementation_service import SspControlImplementationService ``` 在 `OscalContainer` 中新增： ```python # SSP Control Implementation control_implementation_repo = providers.Singleton(ControlImplementationRepoImpl) control_implementation_objective_repo = providers.Singleton(ControlImplementationObjectiveRepoImpl) control_implementation_domain_service = providers.Factory( ControlImplementationDomainService, control_implementation_repo=control_implementation_repo, ) control_implementation_objective_domain_service = providers.Factory( ControlImplementationObjectiveDomainService, repo=control_implementation_objective_repo, ) ssp_control_implementation_service = providers.Factory( SspControlImplementationService, ssp_domain_service=ssp_domain_service, control_impl_domain_service=control_implementation_domain_service, objective_domain_service=control_implementation_objective_domain_service, ) ``` - [ ] **Step 3: 修改 Blueprint 註冊 `api/oscal/__init__.py`** 在 `create_module()` 中新增 imports 與路由註冊： ```python from api.oscal.routes.ssp.ssp_control_implementation_route import ( SspControlImplementationDetailRoute, SspControlImplementationListRoute, SspObjectiveDetailRoute, SspObjectiveDocumentRoute, ) # SSP Control Implementation api.add_resource( SspControlImplementationListRoute, '/ssp//control-implementations', ) api.add_resource( SspControlImplementationDetailRoute, '/ssp//control-implementation/', ) api.add_resource( SspObjectiveDetailRoute, '/ssp//control-implementation//objective/', ) api.add_resource( SspObjectiveDocumentRoute, '/ssp//control-implementation//objective//document/', ) ``` - [ ] **Step 4: 確認 `config/di_modules.py` 的 auto-scan pattern 包含新 route** 新 route 路徑 `api/oscal/routes/ssp/ssp_control_implementation_route.py` 應符合 `api/**/routes/*_route.py` pattern，需確認不在 `EXCLUDE_MODULES` 中。 --- ## Task 10: 專案啟動流程 — Step 7.5 初始化（主專案） **Files:** - Modify: `compliance-manager-be/app/project/service/oscal_project_service.py` - [ ] **Step 1: 在 `OscalProjectService.__init__` 中注入新 service** 新增參數： ```python control_impl_domain_service: ControlImplementationDomainService, objective_domain_service: ControlImplementationObjectiveDomainService, ``` 並存為 `self._ctrl_impl_ds` 和 `self._obj_ds`。 - [ ] **Step 2: 在 DI container 中更新 `OscalProjectService` 的 wiring** 在 `di_containers/project/project_containers.py`（或 `oscal_project_service` 的 DI 配置處）加入： ```python control_impl_domain_service=oscal_container.control_implementation_domain_service, objective_domain_service=oscal_container.control_implementation_objective_domain_service, ``` - [ ] **Step 3: 在 `start_oscal_project()` 中，Step 7 之後新增 Step 7.5** 在 `# 重新抓一次 AP` (line ~244) 之後： ```python # Step 7.5：初始化 SSP Control Implementations + Objectives if ap and ssp: for group in (ap.groups or []): for control in (group.controls or []): ctrl_entity = ControlImplementationEntity( system_security_plan_id=ssp.id, control_identifier=control.control_id, catalog_control_id=getattr(control, "catalog_control_id", None), implementation_status="unknown", # 不可為 None，ORM 欄位 nullable=False created_user=curr_user, updated_user=curr_user, ) ctrl_entity = self._ctrl_impl_ds.add(ctrl_entity) # 重新取得帶 objectives 用的 ctrl map all_ctrls = self._ctrl_impl_ds.get_all_by_ssp_id(ssp.id) ctrl_map = {c.control_identifier: c for c in all_ctrls} for task in (ap.tasks or []): # 找到 task 對應的 control_identifier # task 透過 assessment_task_controls 關聯到 ap_control for task_ctrl in (getattr(task, "controls", []) or []): ctrl_id_str = getattr(task_ctrl, "control_id", None) if not ctrl_id_str: continue parent_ctrl = ctrl_map.get(ctrl_id_str) if not parent_ctrl: continue obj_entity = ControlImplementationObjectiveEntity( system_security_plan_id=ssp.id, control_implementation_id=parent_ctrl.id, control_identifier=ctrl_id_str, statement_identifier=task.task_code or task.title or str(task.id), implementation_status=None, created_user=curr_user, updated_user=curr_user, ) self._obj_ds.add(obj_entity) ``` > **注意：** AP task 與 control 的關聯結構取決於 `init_ap_controls_from_profile` 的實際回傳。上面的 `task.controls` 可能需要根據實際 AP task entity 的屬性調整。實作時須先確認 `ap.tasks[*]` 的資料結構，特別是如何從 task 反查所屬的 `control_id`。可能需要改用 `assessment_task_controls` mapping 表來查詢。 --- ## Task 11: 安裝 jedi-oscal 更新 + 整合測試 - [ ] **Step 1: 在 jedi-oscal 套件中執行安裝** ```bash cd ~/Projects/Jedicogy/module/jedi-python-package/jedi-oscal pip install -e . ``` - [ ] **Step 2: 在主專案執行 migration SQL** ```bash psql -h $DB_HOST -U $DB_USER -d compliance_manager -f scripts/sql/ssp_control_implementation_migration.sql ``` - [ ] **Step 3: 啟動 server 驗證 import 無報錯** ```bash python main_app.py ``` 確認啟動無 ImportError。 - [ ] **Step 4: 手動測試 API（用 curl 或 Swagger）** ```bash # 取得 control implementations 列表 curl -H "Authorization: Bearer $TOKEN" \ http://localhost:8000/api/1.0/ssp/{ssp_uid}/control-implementations # 更新控制項描述 curl -X PUT -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" \ -d '{"implementation_status":"implemented","implementation_description":"test"}' \ http://localhost:8000/api/1.0/ssp/{ssp_uid}/control-implementation/AC.L1-3.1.1 # 更新 AO 描述 curl -X PUT -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" \ -d '{"implementation_description":"AO test"}' \ http://localhost:8000/api/1.0/ssp/{ssp_uid}/control-implementation/AC.L1-3.1.1/objective/[a] ``` --- ## Task 12: GRC Controls List API 回傳 SSP 現況資訊 **需求：** `/grc/project//control-group//controls/list` 回傳每個控制項和 AO 的 SSP 實作現況。 **Files:** - Modify: `infra/grc/repository/grc_control_repo_impl.py` — LEFT JOIN SSP 表取得現況 - Modify: `domain/grc/entities/grc_control_entity.py` — 加 SSP 欄位 - Modify: `infra/grc/mapper/grc_control_mapper.py` — 加 SSP 映射 - Modify: `app/grc/dto/control_dto.py` — 加 SSP 欄位 - Modify: `api/grc/serializers/control.py` — 加 SSP response 欄位 ### 12.1 Entity 加欄位 - [ ] **Step 1: 修改 `GrcControlEntity`** 在 `domain/grc/entities/grc_control_entity.py` 的 `__init__` 新增： ```python # SSP 現況 ssp_implementation_status: str = None, ssp_implementation_description: str = None, ssp_has_description: bool = False, ``` 並在 body 中： ```python self.ssp_implementation_status = ssp_implementation_status self.ssp_implementation_description = ssp_implementation_description self.ssp_has_description = ssp_has_description ``` - [ ] **Step 2: 修改 `GrcAoSummaryEntity`** 在同一檔案的 `GrcAoSummaryEntity.__init__` 新增： ```python ssp_implementation_status: str = None, ssp_implementation_description: str = None, ssp_has_description: bool = False, ``` ### 12.2 Repo 查詢加 LEFT JOIN - [ ] **Step 3: 修改 `GrcControlRepoImpl.list_controls()`** 在 `infra/grc/repository/grc_control_repo_impl.py` 中： **3a) 新增 import：** ```python from jedi_oscal.infra.model.ssp.ssp_control_implementation import OscalSystemSecurityPlanControlImplementation from jedi_oscal.infra.model.ssp.ssp_control_implementation_objective import OscalSspControlImplementationObjective ``` **3b) 在 Step 1（resolve AP ID）之後，查詢 SSP ID：** 已有 `assessment_plan_id`，需透過 AP 的 `ssp_id` 取得 SSP： ```python from jedi_oscal.infra.model.ap.assessment_plan import OscalAssessmentPlan ap_row = session.query(OscalAssessmentPlan.ssp_id).filter( OscalAssessmentPlan.id == assessment_plan_id ).first() ssp_id = ap_row.ssp_id if ap_row else None ``` > 注意：`OscalAssessmentPlan.ssp_id` 是 soft ref（String），需透過 SSP uid → id 轉換， > 或直接用 `system_security_plan_control_implementations.system_security_plan_id` JOIN。 > 實作時確認 AP model 的 `ssp_id` 欄位型別。 **3c) 在 main query 中加 LEFT JOIN SSP control implementation：** ```python # SSP 現況 subquery ssp_impl_status = session.query( OscalSystemSecurityPlanControlImplementation.control_identifier, OscalSystemSecurityPlanControlImplementation.implementation_status, OscalSystemSecurityPlanControlImplementation.implementation_description, ).filter( OscalSystemSecurityPlanControlImplementation.system_security_plan_id == ssp_id, ).subquery("ssp_impl") # 在 main query 加 outerjoin query = query.outerjoin( ssp_impl_status, OscalAssessmentPlanControl.control_id == ssp_impl_status.c.control_identifier, ).add_columns( ssp_impl_status.c.implementation_status.label("ssp_implementation_status"), ssp_impl_status.c.implementation_description.label("ssp_implementation_description"), ) ``` **3d) 在 entity mapping 時傳入 SSP 欄位：** ```python ssp_implementation_status=getattr(row, "ssp_implementation_status", None), ssp_implementation_description=getattr(row, "ssp_implementation_description", None), ssp_has_description=bool(getattr(row, "ssp_implementation_description", None)), ``` - [ ] **Step 4: 在 AO summary 批次查詢中加 SSP objective 資料** 在 Step 9（batch fetch AO summaries）中，加一個批次查詢： ```python # 批次取得 SSP objectives if ssp_id: ssp_objectives = session.query( OscalSspControlImplementationObjective.statement_identifier, OscalSspControlImplementationObjective.control_identifier, OscalSspControlImplementationObjective.implementation_status, OscalSspControlImplementationObjective.implementation_description, ).filter( OscalSspControlImplementationObjective.system_security_plan_id == ssp_id, ).all() # 建立 lookup: (control_identifier, statement_identifier) → obj ssp_obj_map = {} for obj in ssp_objectives: key = (obj.control_identifier, obj.statement_identifier) ssp_obj_map[key] = obj ``` 在建立 `GrcAoSummaryEntity` 時，用 `task.task_code` 作為 `statement_identifier` 查詢： ```python ssp_obj = ssp_obj_map.get((control_identifier, task.task_code)) ao_entity.ssp_implementation_status = ssp_obj.implementation_status if ssp_obj else None ao_entity.ssp_implementation_description = ssp_obj.implementation_description if ssp_obj else None ao_entity.ssp_has_description = bool(ssp_obj.implementation_description) if ssp_obj else False ``` ### 12.3 Mapper 更新 - [ ] **Step 5: 修改 `GrcControlMapper.to_full_entity()`** 在 `infra/grc/mapper/grc_control_mapper.py` 的 `to_full_entity()` 中加入 SSP 參數傳遞。 ### 12.4 DTO 更新 - [ ] **Step 6: 修改 `GrcControlDto`** 在 `app/grc/dto/control_dto.py` 的 `GrcControlDto` 新增： ```python ssp_implementation_status: Optional[str] = None ssp_implementation_description: Optional[str] = None ssp_has_description: bool = False ``` 在 `from_entity()` 中加入映射。 - [ ] **Step 7: 修改 `GrcAoSummaryDto`** ```python ssp_implementation_status: Optional[str] = None ssp_implementation_description: Optional[str] = None ssp_has_description: bool = False ``` ### 12.5 Serializer 更新 - [ ] **Step 8: 修改 `ControlResponseSchema`** 在 `api/grc/serializers/control.py` 的 `ControlResponseSchema` 新增： ```python ssp_implementation_status = fields.Str(allow_none=True) ssp_implementation_description = fields.Str(allow_none=True) ssp_has_description = fields.Bool(dump_default=False) ``` - [ ] **Step 9: 修改 `AoSummarySchema`** ```python ssp_implementation_status = fields.Str(allow_none=True) ssp_implementation_description = fields.Str(allow_none=True) ssp_has_description = fields.Bool(dump_default=False) ``` --- ## 重要決策記錄 | 決策 | 理由 | |------|------| | Objective 新表放 jedi-oscal 而非主專案 | SSP 是 OSCAL 核心概念，應與 catalog/profile/ap 同套件 | | control_identifier 用 VARCHAR(100) 而非 FK | 規格要求 soft reference，避免跨 Round AP 更新影響 SSP | | implementation_status 允許 NULL | 表示「尚未填寫」，區別於任何有效狀態 | | reference_documents 用 JSONB 而非關聯表 | 文件數量少（通常 1-3 筆），JSONB 更簡潔 | | 文件上傳 API（2.4）暫不在此 plan 完整實作 | 需整合 jedi-file-upload，可作為獨立 Task 後續擴充 | | UniqueConstraint 改用 control_identifier | 原 catalog_control_id 改 nullable 後不適合做 unique key |