trestle.tasks.xlsx_to_oscal_poam

class PoamBuilder:
    """Builder class for constructing OSCAL POAM objects from spreadsheet data."""

    def __init__(self, timestamp: str, validator: PoamValidator) -> None:
        """
        Initialize builder.

        Args:
            timestamp: ISO timestamp for metadata
            validator: PoamValidator instance
        """
        self._timestamp = timestamp
        self._validator = validator

    def create_poam_item(self, poam_id: str, row_data: Dict[str, Any]) -> PoamItem:
        """
        Create PoamItem from row data.

        Args:
            poam_id: POAM ID
            row_data: Row data dictionary

        Returns:
            PoamItem object
        """
        # Required fields
        title = _safe_strip(row_data.get(PoamXlsxHelper.WEAKNESS_NAME, ''))
        description = _safe_strip(row_data.get(PoamXlsxHelper.WEAKNESS_DESCRIPTION, ''))

        # Optional fields
        comments = row_data.get(PoamXlsxHelper.COMMENTS)
        if comments and isinstance(comments, str):
            comments = comments.strip() or None

        # Build properties
        props = []

        # Add POAM ID as property (clean it first)
        clean_poam_id = _safe_strip(poam_id) if poam_id else poam_id
        if clean_poam_id:
            props.append(
                Property(name='poam-id', value=clean_poam_id, uuid=None, ns=None, **{'class': None}, group=None)
            )

        # Add control IDs as properties
        controls_str = row_data.get(PoamXlsxHelper.CONTROLS, '')
        if controls_str:
            controls = self._validator.parse_controls(controls_str)
            for ctrl_id in controls:
                props.append(
                    Property(name='control-id', value=ctrl_id, uuid=None, ns=None, **{'class': None}, group=None)
                )

        # Create PoamItem
        poam_item = PoamItem(
            uuid=UUIDManager.poam_item_uuid(poam_id),
            title=title,
            description=description,
            props=props or None,
            links=None,
            origins=None,
            **{'related-findings': None, 'related-observations': None, 'related-risks': None},
        )

        # Add remarks if present
        if comments:
            poam_item.remarks = comments

        return poam_item

    def create_observation(self, poam_id: str, row_data: Dict[str, Any], helper: PoamXlsxHelper) -> Observation:
        """
        Create Observation from row data.

        Args:
            poam_id: POAM ID
            row_data: Row data dictionary
            helper: PoamXlsxHelper instance

        Returns:
            Observation object
        """
        weakness_name = _safe_strip(row_data.get(PoamXlsxHelper.WEAKNESS_NAME, ''))

        # Description is required
        description = f'Weakness detected: {weakness_name}'
        weakness_source_id = row_data.get(PoamXlsxHelper.WEAKNESS_SOURCE_IDENTIFIER)
        if weakness_source_id and isinstance(weakness_source_id, str):
            weakness_source_id = weakness_source_id.strip()
            if weakness_source_id:
                description += f' (Source: {weakness_source_id})'

        # Collected date (required)
        detection_date_value = row_data.get(PoamXlsxHelper.ORIGINAL_DETECTION_DATE)
        collected = helper.parse_date(detection_date_value)
        if collected is None:
            # Default to current timestamp if no date provided
            collected = datetime.datetime.fromisoformat(self._timestamp)

        # Methods are required
        methods = ['TEST']

        # Origins are optional
        origins = None
        detector_source = row_data.get(PoamXlsxHelper.WEAKNESS_DETECTOR_SOURCE)
        if detector_source:
            actor = OriginActor(type='tool', **{'actor-uuid': UUIDManager.actor_uuid(detector_source)})
            origin = Origin(actors=[actor], **{'related-tasks': None})
            origins = [origin]

        # Subjects are optional
        subjects = None
        asset_id = row_data.get(PoamXlsxHelper.ASSET_IDENTIFIER)
        if asset_id:
            subject = SubjectReference(**{'subject-uuid': UUIDManager.actor_uuid(asset_id)}, type='component')
            subjects = [subject]

        observation = Observation(
            uuid=UUIDManager.observation_uuid(poam_id),
            description=description,
            methods=methods,
            collected=collected,
            origins=origins,
            subjects=subjects,
            title=None,
            props=None,
            links=None,
            types=None,
            expires=None,
            **{'relevant-evidence': None},
        )

        return observation

    def create_risk(self, poam_id: str, row_data: Dict[str, Any], helper: PoamXlsxHelper) -> Risk:
        """
        Create Risk from row data.

        Args:
            poam_id: POAM ID
            row_data: Row data dictionary
            helper: PoamXlsxHelper instance

        Returns:
            Risk object
        """
        # Required fields - clean all text fields
        title = _safe_strip(row_data.get(PoamXlsxHelper.WEAKNESS_NAME, ''))
        description = _safe_strip(row_data.get(PoamXlsxHelper.WEAKNESS_DESCRIPTION, ''))
        statement_raw = row_data.get(PoamXlsxHelper.OVERALL_REMEDIATION_PLAN, description)
        # Convert to string and strip, handling both string and non-string values
        if statement_raw:
            statement = str(statement_raw).strip() if isinstance(statement_raw, str) else str(statement_raw)
        else:
            statement = description
        status = RiskStatus(__root__='open')  # Default status for Open POA&M Items sheet

        # Properties
        props = []

        # Helper function to clean and validate property values
        def add_property_if_valid(name: str, value: Any) -> None:
            """Add property if value is valid (non-empty after stripping)."""
            if value is None:
                return

            if isinstance(value, str):
                # Strip all whitespace including newlines, tabs, etc.
                cleaned = value.strip()
                # Additional check: ensure not just whitespace and matches OSCAL pattern
                if cleaned and not cleaned.isspace():
                    try:
                        props.append(
                            Property(name=name, value=cleaned, uuid=None, ns=None, **{'class': None}, group=None)
                        )
                    except Exception as e:
                        logger.warning(f'Could not create property {name} with value "{cleaned[:50]}...": {e}')
            elif value:
                # Non-string value, convert to string
                props.append(Property(name=name, value=str(value), uuid=None, ns=None, **{'class': None}, group=None))

        # Risk ratings as properties
        add_property_if_valid('original-risk-rating', row_data.get(PoamXlsxHelper.ORIGINAL_RISK_RATING))
        add_property_if_valid('adjusted-risk-rating', row_data.get(PoamXlsxHelper.ADJUSTED_RISK_RATING))
        add_property_if_valid('risk-adjustment', row_data.get(PoamXlsxHelper.RISK_ADJUSTMENT))
        add_property_if_valid('false-positive', row_data.get(PoamXlsxHelper.FALSE_POSITIVE))
        add_property_if_valid('operational-requirement', row_data.get(PoamXlsxHelper.OPERATIONAL_REQUIREMENT))
        add_property_if_valid('deviation-rationale', row_data.get(PoamXlsxHelper.DEVIATION_RATIONALE))

        # Deadline is optional
        deadline = None
        completion_date_value = row_data.get(PoamXlsxHelper.SCHEDULED_COMPLETION_DATE)
        if completion_date_value:
            deadline = helper.parse_date(completion_date_value)

        # Remediations with milestones (optional)
        remediations = None
        milestones_str = row_data.get(PoamXlsxHelper.PLANNED_MILESTONES)
        if milestones_str:
            milestones = helper.parse_milestones(milestones_str)
            if milestones:
                tasks = self._create_milestone_tasks(poam_id, milestones, helper)
                remediation = Response(
                    uuid=UUIDManager.remediation_uuid(poam_id),
                    lifecycle='planned',
                    title=f'Remediation for {poam_id}',
                    description=statement,
                    tasks=tasks or None,
                    props=None,
                    links=None,
                    origins=None,
                    **{'required-assets': None},
                )
                remediations = [remediation]

        risk = Risk(
            uuid=UUIDManager.risk_uuid(poam_id),
            title=title,
            description=description,
            statement=statement,
            status=status,
            props=props or None,
            deadline=deadline,
            remediations=remediations,
            links=None,
            origins=None,
            **{
                'threat-ids': None,
                'characterizations': None,
                'mitigating-factors': None,
                'risk-log': None,
                'related-observations': None,
            },
        )

        return risk

    def _create_milestone_tasks(
        self, poam_id: str, milestones: List[Dict[str, Any]], helper: PoamXlsxHelper
    ) -> List[OscalTask]:
        """
        Create OSCAL Task objects from milestone data.

        Args:
            poam_id: POAM ID
            milestones: List of milestone dictionaries
            helper: PoamXlsxHelper instance

        Returns:
            List of OscalTask objects
        """
        tasks = []

        for idx, milestone in enumerate(milestones):
            title = milestone.get('title', '')
            description = milestone.get('description')

            # Timing is optional
            timing = None
            date_str = milestone.get('timing')
            if date_str:
                try:
                    end_date = helper.parse_date(date_str)
                    if end_date:
                        # Create a date range (start = now, end = milestone date)
                        start_date = datetime.datetime.fromisoformat(self._timestamp)
                        timing = Timing(**{'within-date-range': WithinDateRange(start=start_date, end=end_date)})
                except Exception as e:
                    logger.warning(f'Could not parse milestone date "{date_str}": {e}')

            task = OscalTask(
                uuid=UUIDManager.task_uuid(poam_id, idx),
                type='milestone',
                title=title,
                description=description,
                timing=timing,
                props=None,
                links=None,
                dependencies=None,
                subjects=None,
                **{'associated-activities': None, 'responsible-roles': None},
            )
            tasks.append(task)

        return tasks

    def link_objects(self, poam_item: PoamItem, observation: Observation, risk: Risk) -> None:
        """
        Link POAM objects together via UUID references.

        Args:
            poam_item: PoamItem to link
            observation: Observation to link
            risk: Risk to link
        """
        # Link PoamItem to Observation
        poam_item.related_observations = [RelatedObservation(**{'observation-uuid': observation.uuid})]

        # Link PoamItem to Risk
        poam_item.related_risks = [AssociatedRisk(**{'risk-uuid': risk.uuid})]

        # Link Risk to Observation
        risk.related_observations = [RelatedObservation(**{'observation-uuid': observation.uuid})]

class PoamValidator:
    """Validate POAM spreadsheet data before transformation."""

    VALID_RISK_RATINGS = ['Low', 'Moderate', 'High', 'N/A']
    VALID_YES_NO_PENDING = ['Yes', 'No', 'Pending']
    CONTROL_PATTERN = re.compile(r'^[A-Z]{2}-\d+(\(\d+\))?$')

    def __init__(self, validate_mode: str = 'warn') -> None:
        """
        Initialize validator.

        Args:
            validate_mode: 'on' (fail on error), 'warn' (log warnings), or 'off' (skip validation)
        """
        self.validate_mode = validate_mode
        self.errors: List[str] = []
        self.warnings: List[str] = []

    def validate_row(self, row_num: int, row_data: Dict[str, Any]) -> List[str]:
        """
        Validate a single row of data.

        Args:
            row_num: Row number in spreadsheet
            row_data: Dictionary of column_name -> value

        Returns:
            List of validation error messages
        """
        errors = []

        # Required fields
        if not row_data.get('POAM ID'):
            errors.append(f'Row {row_num}: Missing required field "POAM ID"')
        if not row_data.get('Weakness Name'):
            errors.append(f'Row {row_num}: Missing required field "Weakness Name"')
        if not row_data.get('Weakness Description'):
            errors.append(f'Row {row_num}: Missing required field "Weakness Description"')
        if not row_data.get('Controls'):
            errors.append(f'Row {row_num}: Missing required field "Controls"')

        # Risk rating validation
        for rating_field in ['Original Risk Rating', 'Adjusted Risk Rating']:
            value = row_data.get(rating_field, '')
            if value and value not in self.VALID_RISK_RATINGS:
                errors.append(
                    f'Row {row_num}: Invalid {rating_field}: "{value}". '
                    f'Must be one of: {", ".join(self.VALID_RISK_RATINGS)}'
                )

        # Yes/No/Pending field validation
        for field in ['Risk Adjustment', 'False Positive', 'Operational Requirement']:
            value = row_data.get(field, '')
            if value and value not in self.VALID_YES_NO_PENDING:
                errors.append(
                    f'Row {row_num}: Invalid {field}: "{value}". '
                    f'Must be one of: {", ".join(self.VALID_YES_NO_PENDING[:-1])}'
                )

        self.errors.extend(errors)
        return errors

    def parse_controls(self, controls_str: str) -> List[str]:
        """
        Parse and validate control IDs.

        Args:
            controls_str: Comma/space-separated control IDs like "AC-1, AC-2, SC-7(5)"

        Returns:
            List of validated control IDs
        """
        if not controls_str:
            return []

        # Split by comma and/or space
        controls = re.split(r'[,\s]+', controls_str.strip())

        validated = []
        for ctrl in controls:
            ctrl = ctrl.strip()
            if not ctrl:
                continue
            if self.CONTROL_PATTERN.match(ctrl.upper()):
                validated.append(ctrl.upper())
            else:
                if self.validate_mode != 'off':
                    logger.warning(f'Invalid control format: "{ctrl}" (expected format: XX-N or XX-N(N))')

        return validated

    def log_validation_results(self) -> bool:
        """
        Log validation results based on validation mode.

        Returns:
            True if validation passed or mode is 'warn'/'off', False if errors in 'on' mode
        """
        if self.validate_mode == 'off':
            return True

        if self.errors:
            if self.validate_mode == 'on':
                for error in self.errors:
                    logger.error(error)
                return False
            else:  # warn mode
                for error in self.errors:
                    logger.warning(error)

        if self.warnings:
            for warning in self.warnings:
                logger.warning(warning)

        return True

class PoamXlsxHelper:
    """Helper class for reading POAM spreadsheet templates."""

    # Column name constants
    POAM_ID = 'POAM ID'
    CONTROLS = 'Controls'
    WEAKNESS_NAME = 'Weakness Name'
    WEAKNESS_DESCRIPTION = 'Weakness Description'
    WEAKNESS_DETECTOR_SOURCE = 'Weakness Detector Source'
    WEAKNESS_SOURCE_IDENTIFIER = 'Weakness Source Identifier'
    ASSET_IDENTIFIER = 'Asset Identifier'
    POINT_OF_CONTACT = 'Point of Contact'
    RESOURCES_REQUIRED = 'Resources Required'
    OVERALL_REMEDIATION_PLAN = 'Overall Remediation Plan'
    ORIGINAL_DETECTION_DATE = 'Original Detection Date'
    SCHEDULED_COMPLETION_DATE = 'Scheduled Completion Date'
    PLANNED_MILESTONES = 'Planned Milestones'
    MILESTONE_CHANGES = 'Milestone Changes'
    STATUS_DATE = 'Status Date'
    VENDOR_DEPENDENCY = 'Vendor Dependency'
    LAST_VENDOR_CHECKIN_DATE = 'Last Vendor Check-in Date'
    VENDOR_DEPENDENT_PRODUCT_NAME = 'Vendor Dependent Product Name'
    ORIGINAL_RISK_RATING = 'Original Risk Rating'
    ADJUSTED_RISK_RATING = 'Adjusted Risk Rating'
    RISK_ADJUSTMENT = 'Risk Adjustment'
    FALSE_POSITIVE = 'False Positive'
    OPERATIONAL_REQUIREMENT = 'Operational Requirement'
    DEVIATION_RATIONALE = 'Deviation Rationale'
    SUPPORTING_DOCUMENTS = 'Supporting Documents'
    COMMENTS = 'Comments'
    AUTO_APPROVE = 'Auto-Approve'
    BOD_22_01_TRACKING = 'Binding Operational Directive 22-01 tracking'
    BOD_22_01_DUE_DATE = 'Binding Operational Directive 22-01 Due Date'
    CVE = 'CVE'
    SERVICE_NAME = 'Service Name'

    def __init__(self) -> None:
        """Initialize helper."""
        self._column_map: Dict[str, int] = {}
        self._work_sheet = None
        self._header_row = 5  # Template has headers at row 5 (1-indexed)

    def load(self, xlsx_path: pathlib.Path, sheet_name: str = 'Open POA&M Items') -> None:
        """
        Load spreadsheet file and map columns.

        Args:
            xlsx_path: Path to spreadsheet file
            sheet_name: Name of worksheet to load

        Raises:
            FileNotFoundError: If file doesn't exist
            KeyError: If worksheet doesn't exist
        """
        if not xlsx_path.exists():
            raise FileNotFoundError(f'Spreadsheet file not found: {xlsx_path}')

        workbook = load_workbook(filename=str(xlsx_path), data_only=True)

        if sheet_name not in workbook.sheetnames:
            available = ', '.join(workbook.sheetnames)
            raise KeyError(f'Worksheet "{sheet_name}" not found. Available sheets: {available}')

        self._work_sheet = workbook[sheet_name]
        self._map_columns()

    def _map_columns(self) -> None:
        """Map column names to column indices from header row."""
        if self._work_sheet is None:
            return

        # Read header row (row 5 in template, 1-indexed)
        for cell in self._work_sheet[self._header_row]:
            if cell.value and isinstance(cell.value, str):
                col_name = cell.value.strip()
                if col_name:
                    self._column_map[col_name] = cell.column

        logger.debug(f'Mapped {len(self._column_map)} columns')
        if not self._column_map:
            logger.warning(
                f'No columns mapped from row {self._header_row}. '
                'Verify the spreadsheet uses the expected FedRAMP POAM template format.'
            )

    def row_generator(self) -> Iterator[Tuple[int, Dict[str, Any]]]:
        """
        Generate row numbers and data dictionaries for data rows.

        Yields:
            Tuple of (row_number, row_data_dict)
        """
        if self._work_sheet is None:
            return

        # Data starts at row 6 (after header at row 5)
        data_start_row = self._header_row + 1
        max_row = self._work_sheet.max_row

        for row_num in range(data_start_row, max_row + 1):
            row_data = self._get_row_data(row_num)

            # Skip empty rows (no POAM ID)
            if not row_data.get(self.POAM_ID):
                continue

            yield row_num, row_data

    def _get_row_data(self, row_num: int) -> Dict[str, Any]:
        """
        Extract data from a row as dictionary.

        Args:
            row_num: Row number (1-indexed)

        Returns:
            Dictionary mapping column names to cell values
        """
        row_data = {}

        for col_name, col_idx in self._column_map.items():
            cell = self._work_sheet.cell(row=row_num, column=col_idx)
            value = None if isinstance(cell, MergedCell) else cell.value
            row_data[col_name] = self._clean_value(value)

        return row_data

    def _clean_value(self, value: Any) -> Any:
        """
        Clean cell value.

        Args:
            value: Raw cell value

        Returns:
            Cleaned value (strings are stripped, None for empty)
        """
        if value is None:
            return None
        if isinstance(value, str):
            value = value.strip()
            return value if value else None
        return value

    def parse_date(self, date_value: Any) -> Optional[datetime.datetime]:
        """
        Parse spreadsheet date to datetime with timezone.

        Args:
            date_value: Spreadsheet date (datetime object or string)

        Returns:
            datetime with UTC timezone or None
        """
        if date_value is None:
            return None

        if isinstance(date_value, datetime.datetime):
            # Add timezone if missing
            if date_value.tzinfo is None:
                return date_value.replace(tzinfo=datetime.timezone.utc)
            return date_value

        if isinstance(date_value, datetime.date):
            # Convert date to datetime
            dt = datetime.datetime.combine(date_value, datetime.time.min)
            return dt.replace(tzinfo=datetime.timezone.utc)

        if isinstance(date_value, str):
            # Try to parse ISO 8601 format
            try:
                dt = datetime.datetime.fromisoformat(date_value.replace('Z', '+00:00'))
                if dt.tzinfo is None:
                    dt = dt.replace(tzinfo=datetime.timezone.utc)
                return dt
            except (ValueError, AttributeError):
                logger.warning(f'Could not parse date string: "{date_value}"')
                return None

        logger.warning(f'Unexpected date type: {type(date_value)}')
        return None

    def parse_milestones(self, milestones_str: str) -> List[Dict[str, Any]]:
        """
        Parse milestone text into structured format.

        Args:
            milestones_str: Milestone text (may contain multiple milestones)

        Returns:
            List of milestone dictionaries with 'title', 'description', optional 'timing'
        """
        if not milestones_str:
            return []

        milestones = []
        lines = milestones_str.split('\n')

        for line in lines:
            line = line.strip()
            if not line:
                continue

            # Try to parse: "Milestone N: Description [by YYYY-MM-DD]"
            # Use a safer approach: first extract the date if present, then parse the rest
            # This avoids catastrophic backtracking from the original pattern
            date_str = None
            main_line = line

            # Check if line ends with a date pattern and extract it
            # Use string methods instead of regex to avoid ReDoS vulnerability
            # Look for " by YYYY-MM-DD" at the end of the line (14 chars: " by 2024-01-15")
            if len(line) >= 14:
                # Check last 14 chars for " by YYYY-MM-DD" pattern (case-insensitive)
                potential_date_part = line[-14:]
                # Check structure: " by " (4 chars) + "YYYY-MM-DD" (10 chars)
                # Indices in potential_date_part: 0-3=" by ", 4-7=YYYY, 8=-, 9-10=MM, 11=-, 12-13=DD
                if (
                    potential_date_part[:4].lower() == ' by '
                    and len(potential_date_part) == 14
                    and potential_date_part[8] == '-'
                    and potential_date_part[11] == '-'
                ):
                    # Validate it's actually a date format
                    year = potential_date_part[4:8]
                    month = potential_date_part[9:11]
                    day = potential_date_part[12:14]
                    if year.isdigit() and month.isdigit() and day.isdigit():
                        date_str = f'{year}-{month}-{day}'
                        main_line = line[:-14].rstrip()

            # Parse milestone prefix using string operations to avoid any regex ReDoS risk
            # This approach is deterministic and has O(n) complexity with no backtracking
            main_line_lower = main_line.lower()
            milestone_num = None
            remainder = ''

            # Check for "Milestone N" or "Milestone N:" or "Milestone N." patterns
            if main_line_lower.startswith('milestone '):
                # Find the end of "Milestone" and skip whitespace
                idx = 9  # len('milestone')
                while idx < len(main_line) and main_line[idx].isspace():
                    idx += 1
                # Now find the end of the number
                num_start = idx
                while idx < len(main_line) and main_line[idx].isdigit():
                    idx += 1
                if idx > num_start:  # Found at least one digit
                    milestone_num = main_line[:idx]
                    remainder = main_line[idx:]
            # Check for "MN" or "MN:" or "MN." patterns
            elif len(main_line) > 1 and main_line_lower[0] == 'm' and main_line[1].isdigit():
                idx = 1
                while idx < len(main_line) and main_line[idx].isdigit():
                    idx += 1
                milestone_num = main_line[:idx]
                remainder = main_line[idx:]

            if milestone_num:
                # Strip optional separator and whitespace from remainder
                description = remainder.lstrip(':. ')
                # If no description after prefix, use the entire line as title
                if description:
                    milestone = {'title': description, 'description': milestone_num.strip()}
                else:
                    milestone = {'title': main_line.strip(), 'description': 'Milestone'}
                if date_str:
                    milestone['timing'] = date_str
                milestones.append(milestone)
            else:
                # Fallback: treat entire line as milestone title
                milestones.append({'title': line, 'description': 'Milestone'})

        return milestones