# Stage 3：POA&M 矯正 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:** 實作 POA&M CRUD（列表/詳情/更新）+ close-round（觸發覆核），銜接 Stage 2 的稽核迴圈。 **Architecture:** 擴充現有 POA&M 基礎建設（Stage 2 已建 Model/Entity/Mapper/Repo），新增 Domain Service + App Service + Serializer + Route。close-round 放在 `OscalAuditService`，與 launch_audit / confirm_audit 並列。 **Tech Stack:** Python 3.11, Flask-RESTful, SQLAlchemy 2.0, dependency-injector, marshmallow, jedi-oscal, jedi-flow-engine --- ## File Structure ### 新增 | 檔案 | 職責 | |------|------| | `scripts/sql/stage3_migration.sql` | DDL：3 個 ALTER TABLE ADD COLUMN | | `domain/grc/service/grc_poam_domain_service.py` | POA&M domain service，委派 repo | | `app/grc/dto/poam_dto.py` | POA&M DTO（列表 + 詳情） | | `app/grc/service/poam_service.py` | POA&M App Service（CRUD + 狀態驗證） | | `api/grc/serializers/poam.py` | Marshmallow Request/Response schemas | | `api/grc/routes/poam_route.py` | 3 個 Flask Resource（List/Detail/Update） | ### 修改 | 檔案 | 變更 | |------|------| | `infra/grc/model/poam_model.py` | 加 3 個 column | | `domain/grc/entities/poam_entity.py` | 加 3 個欄位 | | `infra/grc/mapper/poam_mapper.py` | 映射新欄位 | | `domain/grc/repository/i_poam_repo.py` | 新增 3 個方法簽名 | | `infra/grc/repository/poam_repo_impl.py` | 實作 list/get/update + JOIN | | `common/code/grc_error_code.py` | 新增 5 個 error codes | | `app/project/service/oscal_audit_service.py` | 新增 `close_round` 方法 | | `di_containers/grc/grc_containers.py` | 加 poam_domain_service / poam_service | | `api/grc/routes/audit_route.py` | 新增 CloseRoundResource | | `api/grc/__init__.py` | 註冊 4 個 URL | --- ## Task 1：DB Migration + Error Codes **Files:** - Create: `scripts/sql/stage3_migration.sql` - Modify: `common/code/grc_error_code.py` - [ ] **Step 1: 建立 migration SQL** ```sql -- Stage 3: POA&M Remediation ALTER TABLE compliance.poams ADD COLUMN IF NOT EXISTS remediation_plan TEXT; ALTER TABLE compliance.poams ADD COLUMN IF NOT EXISTS due_date DATE; ALTER TABLE compliance.poams ADD COLUMN IF NOT EXISTS assignee_uid VARCHAR(36); ``` 儲存至 `scripts/sql/stage3_migration.sql`。 - [ ] **Step 2: 新增 error codes** 在 `common/code/grc_error_code.py` 的 class 尾部新增： ```python # ── POA&M ──────────────────────────────────────────────────────────── GRC_POAM_NOT_FOUND = ("POA&M 不存在", "GRC_404019") GRC_POAM_INVALID_STATUS_TRANSITION = ("POA&M 狀態轉換不合法", "GRC_412006") GRC_POAM_REMEDIATION_PLAN_REQUIRED = ("需填寫矯正方案", "GRC_412007") # ── Close Round ────────────────────────────────────────────────────── GRC_AP_NOT_REMEDIATION = ("稽核計畫不在矯正中狀態", "GRC_412004") GRC_POAM_NOT_ALL_CLOSED = ("尚有 POA&M 未結案", "GRC_412005") ``` --- ## Task 2：Model / Entity / Mapper 擴充 **Files:** - Modify: `infra/grc/model/poam_model.py` - Modify: `domain/grc/entities/poam_entity.py` - Modify: `infra/grc/mapper/poam_mapper.py` - [ ] **Step 1: PoamModel 加 3 個 column** 在 `infra/grc/model/poam_model.py` 的 `org_unit_id` 後面新增： ```python remediation_plan: Mapped[Optional[str]] = mapped_column(Text, nullable=True) due_date: Mapped[Optional[datetime]] = mapped_column(Date, nullable=True) assignee_uid: Mapped[Optional[str]] = mapped_column(String(36), nullable=True) ``` 頂部 import 加上 `Text, Date`： ```python from sqlalchemy import String, Integer, DateTime, Text, Date ``` `datetime` import 補上 `date`（如果 `due_date` 用 `Date` type，Mapped 型別用 `date`）： ```python from datetime import datetime, date # due_date 改為： due_date: Mapped[Optional[date]] = mapped_column(Date, nullable=True) ``` - [ ] **Step 2: PoamEntity 加 3 個欄位** 在 `domain/grc/entities/poam_entity.py` 的 `__init__` 中，`closed_at=None` 之後加入： ```python remediation_plan=None, due_date=None, assignee_uid=None, ``` 在 `self.closed_at = closed_at` 之後加入： ```python self.remediation_plan = remediation_plan self.due_date = due_date self.assignee_uid = assignee_uid ``` - [ ] **Step 3: PoamMapper 映射新欄位** 在 `infra/grc/mapper/poam_mapper.py` 的 `to_entity` 中，`closed_at` 之後加入： ```python remediation_plan=model.remediation_plan, due_date=model.due_date, assignee_uid=model.assignee_uid, ``` 在 `to_model` 中，`status=entity.status` 之後加入： ```python remediation_plan=entity.remediation_plan, due_date=entity.due_date, assignee_uid=entity.assignee_uid, ``` --- ## Task 3：Repo 層（Interface + Impl） **Files:** - Modify: `domain/grc/repository/i_poam_repo.py` - Modify: `infra/grc/repository/poam_repo_impl.py` - [ ] **Step 1: IPoamRepo 新增方法簽名** ```python from abc import abstractmethod from typing import List, Optional from jedi_common.interfaces.entities import PageDataEntity from jedi_common.interfaces.spec.base_query_spec import PageSpec, SortSpec from domain.grc.entities.poam_entity import PoamEntity class IPoamRepo: @abstractmethod def batch_create(self, entities: List[PoamEntity]) -> List[PoamEntity]: pass @abstractmethod def list_poams( self, ap_uid: str, page_spec: Optional[PageSpec], sorts: Optional[List[SortSpec]], **filters ) -> PageDataEntity: pass @abstractmethod def get_poam(self, ap_uid: str, poam_uid: str) -> Optional[PoamEntity]: pass @abstractmethod def update_poam(self, poam_uid: str, updates: dict, curr_user: str) -> Optional[PoamEntity]: pass ``` - [ ] **Step 2: PoamRepoImpl 實作查詢** 覆寫整個 `infra/grc/repository/poam_repo_impl.py`： ```python import math import uuid from datetime import datetime from typing import List, Optional from sqlalchemy import func as sa_func from jedi_common.interfaces.entities import PageDataEntity, PageMetaEntity from jedi_common.interfaces.spec.base_query_spec import PageSpec, SortSpec from jedi_common.session.database.session_context import get_session from jedi_oscal.infra.model.ap.assessment_plan import OscalAssessmentPlan from jedi_oscal.infra.model.ar.assessment_result_finding import OscalAssessmentResultFinding from domain.grc.entities.poam_entity import PoamEntity from domain.grc.repository.i_poam_repo import IPoamRepo from infra.grc.model.poam_model import PoamModel from infra.grc.mapper.poam_mapper import PoamMapper class PoamRepoImpl(IPoamRepo): @property def session(self): return get_session() # ── batch_create（Stage 2 原有） ────────────────────────────────────── def batch_create(self, entities: List[PoamEntity]) -> List[PoamEntity]: models = [] for entity in entities: entity.uid = str(uuid.uuid4()) model = PoamMapper.to_model(entity) self.session.add(model) models.append(model) self.session.flush() return [PoamMapper.to_entity(m) for m in models] # ── list_poams ──────────────────────────────────────────────────────── def list_poams( self, ap_uid: str, page_spec: Optional[PageSpec], sorts: Optional[List[SortSpec]], **filters ) -> PageDataEntity: page = page_spec.page if page_spec else 1 page_size = page_spec.page_size if page_spec else 25 # 取得 AP id ap = self.session.query(OscalAssessmentPlan.id).filter( OscalAssessmentPlan.uid == ap_uid ).first() if ap is None: return PageDataEntity( meta=PageMetaEntity(paging=True, page=page, page_size=page_size, total=0, total_pages=0, has_next=False, has_prev=False), data=[], ) # 主查詢：POA&M LEFT JOIN finding（取 title/severity/category） from jedi_auth.infra.models.user import User query = ( self.session.query( PoamModel, OscalAssessmentResultFinding.title.label("finding_title"), OscalAssessmentResultFinding.severity.label("finding_severity"), OscalAssessmentResultFinding.category.label("finding_category"), User.nickname.label("assignee_name"), ) .outerjoin( OscalAssessmentResultFinding, OscalAssessmentResultFinding.id == PoamModel.ar_finding_id, ) .outerjoin( User, User.uid == PoamModel.assignee_uid, ) .filter(PoamModel.assessment_plan_id == ap.id) ) # filters status = filters.get("status") if status: query = query.filter(PoamModel.status == status) search = filters.get("search") if search: like = f"%{search}%" query = query.filter( sa_func.coalesce(PoamModel.control_identifier, "").ilike(like) | sa_func.coalesce(OscalAssessmentResultFinding.title, "").ilike(like) ) # sort if sorts: for s in sorts: col = getattr(PoamModel, s.field, None) if col is not None: query = query.order_by(col.desc() if s.order == "desc" else col.asc()) else: query = query.order_by(PoamModel.created_at.desc()) total = query.count() rows = query.offset((page - 1) * page_size).limit(page_size).all() entities = [] for row in rows: poam_model = row[0] entity = PoamMapper.to_entity(poam_model) entity.finding_title = row.finding_title entity.finding_severity = row.finding_severity entity.finding_category = row.finding_category entity.assignee_name = row.assignee_name entities.append(entity) return PageDataEntity( meta=PageMetaEntity( paging=True, page=page, page_size=page_size, total=total, total_pages=math.ceil(total / page_size) if page_size else 0, has_next=page * page_size < total, has_prev=page > 1, ), data=entities, ) # ── get_poam ────────────────────────────────────────────────────────── def get_poam(self, ap_uid: str, poam_uid: str) -> Optional[PoamEntity]: from jedi_auth.infra.models.user import User ap = self.session.query(OscalAssessmentPlan.id).filter( OscalAssessmentPlan.uid == ap_uid ).first() if ap is None: return None row = ( self.session.query( PoamModel, User.nickname.label("assignee_name"), ) .outerjoin(User, User.uid == PoamModel.assignee_uid) .filter( PoamModel.uid == poam_uid, PoamModel.assessment_plan_id == ap.id, ) .first() ) if row is None: return None entity = PoamMapper.to_entity(row[0]) entity.assignee_name = row.assignee_name return entity # ── update_poam ─────────────────────────────────────────────────────── def update_poam(self, poam_uid: str, updates: dict, curr_user: str) -> Optional[PoamEntity]: model = self.session.query(PoamModel).filter(PoamModel.uid == poam_uid).first() if model is None: return None for key, value in updates.items(): if hasattr(model, key): setattr(model, key, value) model.updated_user = curr_user self.session.flush() return PoamMapper.to_entity(model) ``` --- ## Task 4：Domain Service **Files:** - Create: `domain/grc/service/grc_poam_domain_service.py` - [ ] **Step 1: 建立 GrcPoamDomainService** ```python from typing import Optional, List from jedi_common.interfaces.entities import PageDataEntity from jedi_common.interfaces.spec.base_query_spec import PageSpec, SortSpec from domain.grc.entities.poam_entity import PoamEntity from domain.grc.repository.i_poam_repo import IPoamRepo class GrcPoamDomainService: def __init__(self, poam_repo: IPoamRepo): self._repo = poam_repo def list_poams( self, ap_uid: str, page_spec: Optional[PageSpec], sorts: Optional[List[SortSpec]], **filters ) -> PageDataEntity: return self._repo.list_poams(ap_uid, page_spec, sorts, **filters) def get_poam(self, ap_uid: str, poam_uid: str) -> Optional[PoamEntity]: return self._repo.get_poam(ap_uid, poam_uid) def update_poam(self, poam_uid: str, updates: dict, curr_user: str) -> Optional[PoamEntity]: return self._repo.update_poam(poam_uid, updates, curr_user) def batch_create(self, entities: List[PoamEntity]) -> List[PoamEntity]: return self._repo.batch_create(entities) ``` --- ## Task 5：App 層（DTO + Service） **Files:** - Create: `app/grc/dto/poam_dto.py` - Create: `app/grc/service/poam_service.py` - [ ] **Step 1: 建立 POA&M DTOs** ```python from dataclasses import dataclass, field from typing import Optional from datetime import datetime, date @dataclass class PoamFindingDto: uid: str title: Optional[str] = None category: Optional[str] = None severity: Optional[str] = None description: Optional[str] = None recommendation: Optional[str] = None @staticmethod def from_finding_model(model) -> "PoamFindingDto": return PoamFindingDto( uid=str(model.uid), title=model.title, category=model.category, severity=model.severity, description=model.description, recommendation=model.recommendation, ) @dataclass class PoamJobInfoDto: workflow_execution_id: Optional[int] = None job_uid: Optional[str] = None job_status: Optional[str] = None @dataclass class PoamDto: uid: str control_identifier: Optional[str] = None ao_uid: Optional[str] = None status: Optional[str] = None remediation_plan: Optional[str] = None due_date: Optional[date] = None assignee_uid: Optional[str] = None assignee_name: Optional[str] = None closed_at: Optional[datetime] = None created_at: Optional[datetime] = None updated_at: Optional[datetime] = None # 列表用（JOIN 來的） finding_title: Optional[str] = None finding_severity: Optional[str] = None finding_category: Optional[str] = None @staticmethod def from_entity(entity) -> "PoamDto": return PoamDto( uid=entity.uid, control_identifier=entity.control_identifier, ao_uid=entity.ao_uid, status=entity.status, remediation_plan=entity.remediation_plan, due_date=entity.due_date, assignee_uid=entity.assignee_uid, assignee_name=getattr(entity, "assignee_name", None), closed_at=entity.closed_at, created_at=entity.created_at, updated_at=entity.updated_at, finding_title=getattr(entity, "finding_title", None), finding_severity=getattr(entity, "finding_severity", None), finding_category=getattr(entity, "finding_category", None), ) @staticmethod def from_entity_list(entities) -> list: return [PoamDto.from_entity(e) for e in entities] @dataclass class PoamDetailDto(PoamDto): finding: Optional[PoamFindingDto] = None job_info: Optional[PoamJobInfoDto] = None ``` - [ ] **Step 2: 建立 PoamService** ```python from datetime import datetime from typing import Optional from jedi_common.handler.exception import NotFound, PreconditionFailedError from jedi_common.interfaces.dto import PageDataDto, PageDto from jedi_common.interfaces.spec.base_query_spec import PageSpec, SortSpec from jedi_common.session.database.db import transaction from jedi_common.session.database.session_context import get_session from app.grc.dto.poam_dto import PoamDto, PoamDetailDto, PoamFindingDto, PoamJobInfoDto from common.code.grc_error_code import GrcErrorCode from domain.grc.service.grc_poam_domain_service import GrcPoamDomainService # 合法狀態轉換 _VALID_TRANSITIONS = { ("open", "in_progress"), ("open", "closed"), ("in_progress", "closed"), ("closed", "in_progress"), } class PoamService: def __init__(self, grc_poam_domain_service: GrcPoamDomainService): self._domain_service = grc_poam_domain_service @transaction def list_poams(self, ap_uid: str, pager: dict, sorts: list, **filters) -> PageDataDto: page_spec = PageSpec(**pager) if pager else None sorts_spec = [SortSpec(**s) for s in sorts] if sorts else None result = self._domain_service.list_poams(ap_uid, page_spec, sorts_spec, **filters) return PageDataDto( meta=PageDto.from_entity(result.meta), data=PoamDto.from_entity_list(result.data), ) @transaction def get_poam_detail(self, ap_uid: str, poam_uid: str) -> Optional[PoamDetailDto]: entity = self._domain_service.get_poam(ap_uid, poam_uid) if entity is None: return None base = PoamDto.from_entity(entity) # 查 finding 詳情 finding_dto = None if entity.ar_finding_id: finding_dto = self._load_finding(entity.ar_finding_id) # 查 job info job_info_dto = None if entity.ao_uid: job_info_dto = self._load_job_info(entity.ao_uid) return PoamDetailDto( **{k: v for k, v in base.__dict__.items()}, finding=finding_dto, job_info=job_info_dto, ) @transaction def update_poam(self, ap_uid: str, poam_uid: str, curr_user: str, **data) -> PoamDto: entity = self._domain_service.get_poam(ap_uid, poam_uid) if entity is None: raise NotFound(GrcErrorCode.GRC_POAM_NOT_FOUND) new_status = data.get("status") # 狀態轉換驗證 if new_status and new_status != entity.status: if (entity.status, new_status) not in _VALID_TRANSITIONS: raise PreconditionFailedError(GrcErrorCode.GRC_POAM_INVALID_STATUS_TRANSITION) # open → in_progress 需要 remediation_plan if entity.status == "open" and new_status == "in_progress": plan = data.get("remediation_plan") or entity.remediation_plan if not plan: raise PreconditionFailedError(GrcErrorCode.GRC_POAM_REMEDIATION_PLAN_REQUIRED) # 組 updates dict updates = {} for field in ("status", "remediation_plan", "due_date", "assignee_uid"): if field in data and data[field] is not None: updates[field] = data[field] # 自動寫入 / 清除 closed_at if new_status == "closed": updates["closed_at"] = datetime.now() elif new_status and new_status != "closed" and entity.closed_at: updates["closed_at"] = None updated = self._domain_service.update_poam(poam_uid, updates, curr_user) return PoamDto.from_entity(updated) def _load_finding(self, ar_finding_id: int) -> Optional[PoamFindingDto]: from jedi_oscal.infra.model.ar.assessment_result_finding import OscalAssessmentResultFinding session = get_session() finding = session.query(OscalAssessmentResultFinding).filter( OscalAssessmentResultFinding.id == ar_finding_id ).first() if finding is None: return None return PoamFindingDto.from_finding_model(finding) def _load_job_info(self, ao_uid: str) -> Optional[PoamJobInfoDto]: from jedi_oscal.infra.model.ap.assessment_plan_task import OscalAssessmentPlanTask from infra.project.models.assessment_plan_task_workflow_execution_mapping import ( AssessmentPlanTaskWorkflowExecutionMapping, ) from jedi_flow_engine.infra.models.job_execution import JobExecution session = get_session() task = session.query(OscalAssessmentPlanTask).filter( OscalAssessmentPlanTask.uid == ao_uid ).first() if task is None: return None mapping = session.query(AssessmentPlanTaskWorkflowExecutionMapping).filter( AssessmentPlanTaskWorkflowExecutionMapping.assessment_plan_task_id == task.id ).first() if mapping is None: return None job = ( session.query(JobExecution) .filter( JobExecution.workflow_execution_id == mapping.workflow_execution_id, JobExecution.type == "USER", ) .order_by(JobExecution.id.desc()) .first() ) if job is None: return None return PoamJobInfoDto( workflow_execution_id=mapping.workflow_execution_id, job_uid=job.uid, job_status=job.status, ) ``` --- ## Task 6：API 層（Serializer + Route） **Files:** - Create: `api/grc/serializers/poam.py` - Create: `api/grc/routes/poam_route.py` - [ ] **Step 1: 建立 Serializer** ```python from marshmallow import Schema, fields from jedi_common.interfaces.schema.common import RequestMetaSchema, EnvelopeSchema # ── Filters ── class PoamFiltersSchema(Schema): status = fields.String(load_default=None, allow_none=True) search = fields.String(load_default=None, allow_none=True) # ── Request ── class PoamListRequestSchema(RequestMetaSchema): filters = fields.Nested(PoamFiltersSchema, missing={}, allow_none=True) class PoamUpdateRequestSchema(Schema): status = fields.String(load_default=None, allow_none=True) remediation_plan = fields.String(load_default=None, allow_none=True) due_date = fields.Date(load_default=None, allow_none=True) assignee_uid = fields.String(load_default=None, allow_none=True) # ── Response items ── class PoamResponseSchema(Schema): uid = fields.String() control_identifier = fields.String(allow_none=True) ao_uid = fields.String(allow_none=True) status = fields.String(allow_none=True) remediation_plan = fields.String(allow_none=True) due_date = fields.Date(allow_none=True) assignee_uid = fields.String(allow_none=True) assignee_name = fields.String(allow_none=True) closed_at = fields.DateTime(allow_none=True) created_at = fields.DateTime(allow_none=True) updated_at = fields.DateTime(allow_none=True) # 列表用 finding_title = fields.String(allow_none=True) finding_severity = fields.String(allow_none=True) finding_category = fields.String(allow_none=True) class PoamFindingResponseSchema(Schema): uid = fields.String() title = fields.String(allow_none=True) category = fields.String(allow_none=True) severity = fields.String(allow_none=True) description = fields.String(allow_none=True) recommendation = fields.String(allow_none=True) class PoamJobInfoResponseSchema(Schema): workflow_execution_id = fields.Integer(allow_none=True) job_uid = fields.String(allow_none=True) job_status = fields.String(allow_none=True) class PoamDetailInnerSchema(PoamResponseSchema): finding = fields.Nested(PoamFindingResponseSchema, allow_none=True) job_info = fields.Nested(PoamJobInfoResponseSchema, allow_none=True) # ── Response envelopes ── class PoamListResponseSchema(EnvelopeSchema): data = fields.Nested(PoamResponseSchema, many=True, dump_default=[]) class PoamDetailResponseSchema(Schema): data = fields.Nested(PoamDetailInnerSchema, allow_none=True) # ── Close Round ── class CloseRoundResponseSchema(Schema): ap_uid = fields.String() status = fields.String() reset_count = fields.Integer() ``` - [ ] **Step 2: 建立 Route** ```python import logging 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 api.grc.serializers.poam import ( PoamListRequestSchema, PoamListResponseSchema, PoamUpdateRequestSchema, PoamResponseSchema, PoamDetailResponseSchema, ) from app.grc.service.poam_service import PoamService from common.enum.schema_code import AUTH_PARAMS from common.util.response_util import return_response from di_containers.containers import Containers logger = logging.getLogger(__name__) class PoamListResource(MethodResource): """POST /grc/project//ap//poams/list""" @doc(description="POA&M 分頁列表", tags=["GRC POA&M"], params=AUTH_PARAMS) @use_kwargs(PoamListRequestSchema, location="json", apply=False) @marshal_with(PoamListResponseSchema, apply=False) @jwt_required() @inject def post( self, project_uid: str, ap_uid: str, poam_service: PoamService = Provide[Containers.grc_container.poam_service], ): payload = request.get_json(silent=True) or {} data = PoamListRequestSchema().load(payload) pager = data.get("pager") sorts = data.get("sort") filters = data.get("filters") or {} result = poam_service.list_poams(ap_uid, pager, sorts, **filters) return return_response(True, PoamListResponseSchema().dump(result)) class PoamDetailResource(MethodResource): """GET /grc/project//ap//poam/""" @doc(description="POA&M 詳情（含 finding + job info）", tags=["GRC POA&M"], params=AUTH_PARAMS) @marshal_with(PoamDetailResponseSchema, apply=False) @jwt_required() @inject def get( self, project_uid: str, ap_uid: str, poam_uid: str, poam_service: PoamService = Provide[Containers.grc_container.poam_service], ): result = poam_service.get_poam_detail(ap_uid, poam_uid) if result is None: return return_response(False, None) return return_response(True, PoamDetailResponseSchema().dump({"data": result})) class PoamUpdateResource(MethodResource): """PUT /grc/project//ap//poam/""" @doc(description="更新 POA&M（狀態 / 改善計畫 / 負責人）", tags=["GRC POA&M"], params=AUTH_PARAMS) @use_kwargs(PoamUpdateRequestSchema, location="json", apply=False) @marshal_with(PoamResponseSchema, apply=False) @jwt_required() @inject def put( self, project_uid: str, ap_uid: str, poam_uid: str, poam_service: PoamService = Provide[Containers.grc_container.poam_service], ): payload = request.get_json(silent=True) or {} data = PoamUpdateRequestSchema().load(payload) from jedi_common.session.auth.auth_context import get_user_context user = get_user_context() result = poam_service.update_poam(ap_uid, poam_uid, user.login_name, **data) return return_response(True, PoamResponseSchema().dump(result)) ``` --- ## Task 7：close_round 編排 **Files:** - Modify: `app/project/service/oscal_audit_service.py` - [ ] **Step 1: 在 OscalAuditService 新增 close_round 方法** 在 `_revert_ao_workflow` 方法之前（`confirm_audit` 之後），新增： ```python @transaction def close_round(self, ap_uid: str, curr_user) -> dict: session = get_session() # 1. 查 AP ap = self._ap_service.get_assessment_plan(uid=ap_uid) if ap is None: raise NotFound(GrcErrorCode.GRC_AP_NOT_FOUND) if ap.status != "remediation": raise PreconditionFailedError(GrcErrorCode.GRC_AP_NOT_REMEDIATION) # 2. 查該 AP 所有 POA&M from infra.grc.model.poam_model import PoamModel poams = ( session.query(PoamModel) .filter(PoamModel.assessment_plan_id == ap.id) .all() ) # 3. 驗證全部 closed not_closed = [p for p in poams if p.status != "closed"] if not_closed: from jedi_common.handler.exception import BadRequestError raise BadRequestError(GrcErrorCode.GRC_POAM_NOT_ALL_CLOSED) # 4. 收集需要重置的 control_identifiers control_ids = list({p.control_identifier for p in poams}) # 5. 找 ar_data (run_no=1) 下對應的 ar_controls，重置 verdict ar = self._ar_service.get_assessment_result(assessment_plan_id=ap.id) from jedi_oscal.infra.model.ar.assessment_result_data import OscalAssessmentResultData from jedi_oscal.infra.model.ar.assessment_result_control import OscalAssessmentResultControl ar_data = ( session.query(OscalAssessmentResultData) .filter( OscalAssessmentResultData.assessment_result_id == ar.id, OscalAssessmentResultData.run_no == 1, ) .first() ) reset_count = 0 if ar_data and control_ids: ar_controls = ( session.query(OscalAssessmentResultControl) .filter( OscalAssessmentResultControl.assessment_result_data_id == ar_data.id, OscalAssessmentResultControl.control_id.in_(control_ids), ) .all() ) for arc in ar_controls: arc.verdict = None arc.confidence = None arc.rationale = None arc.remarks = None arc.updated_user = str(curr_user) reset_count = len(ar_controls) session.flush() # 6. AP status → auditing self._ap_service.update_assessment_plan(ap.id, {"status": "auditing"}, curr_user) return {"ap_uid": ap_uid, "status": "auditing", "reset_count": reset_count} ``` 需在檔案頂部確認 import：`GrcErrorCode` 中新增的 `GRC_AP_NOT_REMEDIATION` 和 `GRC_POAM_NOT_ALL_CLOSED` 已在 Task 1 加入。 --- ## Task 8：DI Container + Blueprint 註冊 **Files:** - Modify: `di_containers/grc/grc_containers.py` - Modify: `api/grc/routes/audit_route.py` - Modify: `api/grc/__init__.py` - [ ] **Step 1: GrcContainer 加 POA&M providers** 在 `di_containers/grc/grc_containers.py` 頂部加 import： ```python from app.grc.service.poam_service import PoamService from domain.grc.service.grc_poam_domain_service import GrcPoamDomainService ``` 在 `audit_service` provider 之後加入： ```python # ── POA&M ───────────────────────────────────────────────────────────── grc_poam_domain_service = providers.Factory( GrcPoamDomainService, poam_repo=poam_repo, ) poam_service = providers.Factory( PoamService, grc_poam_domain_service=grc_poam_domain_service, ) ``` > 注意：`poam_repo` 是 Stage 2 已加入的 `providers.Singleton(PoamRepoImpl)`，直接複用。 - [ ] **Step 2: audit_route.py 新增 CloseRoundResource** 在 `api/grc/routes/audit_route.py` 的 import 區加入： ```python from api.grc.serializers.poam import CloseRoundResponseSchema ``` 在 `ConfirmAuditResource` 之後加入： ```python class CloseRoundResource(MethodResource): """POST /grc/project//ap//close-round""" @doc(description="觸發覆核（close-round）", tags=["GRC Audit"], params=AUTH_PARAMS) @marshal_with(CloseRoundResponseSchema, apply=False) @jwt_required() @inject def post( self, project_uid: str, ap_uid: str, oscal_audit_service: OscalAuditService = Provide[ Containers.project_container.oscal_audit_service ], ): user = get_user_context() _check_auditor_role(project_uid, user.id) result = oscal_audit_service.close_round( ap_uid=ap_uid, curr_user=user.login_name, ) return return_response(True, CloseRoundResponseSchema().dump(result)) ``` - [ ] **Step 3: Blueprint 註冊 4 個 URL** 在 `api/grc/__init__.py` 的 import 區加入： ```python from api.grc.routes.poam_route import ( PoamListResource, PoamDetailResource, PoamUpdateResource, ) from api.grc.routes.audit_route import CloseRoundResource ``` 在 `# --- Audit (AP-scoped) ---` 區塊尾部（`ArFindingDetailResource` 之後）加入： ```python # --- POA&M (AP-scoped) --- api.add_resource( PoamListResource, "/project//ap//poams/list", ) api.add_resource( PoamDetailResource, "/project//ap//poam/", ) api.add_resource( PoamUpdateResource, "/project//ap//poam/", endpoint="poam_update", ) api.add_resource( CloseRoundResource, "/project//ap//close-round", ) ``` > **注意**：`PoamDetailResource`（GET）和 `PoamUpdateResource`（PUT）共用同一 URL path。Flask-RESTful 需要透過 `endpoint` 參數區分。或者合併成一個 Resource class 同時處理 GET 和 PUT。 **更好的做法：合併 Detail + Update 為一個 Resource**。回到 `api/grc/routes/poam_route.py`，把 `PoamDetailResource` 和 `PoamUpdateResource` 合併： ```python class PoamDetailResource(MethodResource): """GET/PUT /grc/project//ap//poam/""" @doc(description="POA&M 詳情（含 finding + job info）", tags=["GRC POA&M"], params=AUTH_PARAMS) @marshal_with(PoamDetailResponseSchema, apply=False) @jwt_required() @inject def get( self, project_uid: str, ap_uid: str, poam_uid: str, poam_service: PoamService = Provide[Containers.grc_container.poam_service], ): result = poam_service.get_poam_detail(ap_uid, poam_uid) if result is None: return return_response(False, None) return return_response(True, PoamDetailResponseSchema().dump({"data": result})) @doc(description="更新 POA&M（狀態 / 改善計畫 / 負責人）", tags=["GRC POA&M"], params=AUTH_PARAMS) @use_kwargs(PoamUpdateRequestSchema, location="json", apply=False) @marshal_with(PoamResponseSchema, apply=False) @jwt_required() @inject def put( self, project_uid: str, ap_uid: str, poam_uid: str, poam_service: PoamService = Provide[Containers.grc_container.poam_service], ): payload = request.get_json(silent=True) or {} data = PoamUpdateRequestSchema().load(payload) from jedi_common.session.auth.auth_context import get_user_context user = get_user_context() result = poam_service.update_poam(ap_uid, poam_uid, user.login_name, **data) return return_response(True, PoamResponseSchema().dump(result)) ``` 這樣 Blueprint 註冊就只需要： ```python api.add_resource( PoamDetailResource, "/project//ap//poam/", ) ``` --- ## Task 9：驗證 - [ ] **Step 1: 語法檢查** ```bash python -c " import ast, sys files = [ 'scripts/sql/stage3_migration.sql', # skip (SQL) 'common/code/grc_error_code.py', 'infra/grc/model/poam_model.py', 'domain/grc/entities/poam_entity.py', 'infra/grc/mapper/poam_mapper.py', 'domain/grc/repository/i_poam_repo.py', 'infra/grc/repository/poam_repo_impl.py', 'domain/grc/service/grc_poam_domain_service.py', 'app/grc/dto/poam_dto.py', 'app/grc/service/poam_service.py', 'api/grc/serializers/poam.py', 'api/grc/routes/poam_route.py', 'api/grc/routes/audit_route.py', 'app/project/service/oscal_audit_service.py', 'di_containers/grc/grc_containers.py', 'api/grc/__init__.py', ] for f in files: try: with open(f) as fh: ast.parse(fh.read()) print(f'OK: {f}') except SyntaxError as e: print(f'FAIL: {f} -> {e}') sys.exit(1) print('All files passed syntax check.') " ``` - [ ] **Step 2: Import 驗證** ```bash python -c " from domain.grc.entities.poam_entity import PoamEntity from domain.grc.service.grc_poam_domain_service import GrcPoamDomainService from app.grc.dto.poam_dto import PoamDto, PoamDetailDto, PoamFindingDto, PoamJobInfoDto from api.grc.serializers.poam import ( PoamListRequestSchema, PoamListResponseSchema, PoamUpdateRequestSchema, PoamResponseSchema, PoamDetailResponseSchema, CloseRoundResponseSchema, ) print('All imports OK') " ```