Docker Development

# docker-compose.sample.yml — Compose file with common anti-patterns
#
# This Docker Compose file contains deliberate issues for the Docker
# development skill scanner to detect:
#   - Port conflicts (multiple services on same host port)
#   - Missing health checks
#   - Using "latest" tags
#   - No resource limits
#   - Privileged mode enabled
#   - Bind-mounting sensitive host paths
#   - No restart policy on critical services
#   - Hardcoded secrets in environment
#   - Missing networks isolation

version: "3.8"

services:
  # ANTI-PATTERN: no health check, no resource limits, latest tag
  app:
    build:
      context: .
      dockerfile: Dockerfile
    image: acme/dashboard:latest
    ports:
      - "8080:3000"
      - "9229:9229"   # debug port exposed in production
    environment:
      - NODE_ENV=production
      # ANTI-PATTERN: hardcoded secrets
      - DB_PASSWORD=supersecret123
      - JWT_SECRET=my-jwt-secret-do-not-share
      - REDIS_URL=redis://:redispass@redis:6379
    volumes:
      # ANTI-PATTERN: bind-mounting source code in production
      - .:/app
      - /var/run/docker.sock:/var/run/docker.sock  # security risk
    depends_on:
      - postgres
      - redis
    # ANTI-PATTERN: no restart policy

  # ANTI-PATTERN: port conflict with app debug port
  worker:
    image: acme/worker:latest
    ports:
      - "8080:8080"   # CONFLICT: host port 8080 already used by app
    environment:
      - DB_PASSWORD=supersecret123
      - QUEUE_URL=amqp://guest:guest@rabbitmq:5672
    # ANTI-PATTERN: privileged mode is almost never needed
    privileged: true
    depends_on:
      - rabbitmq

  postgres:
    image: postgres:latest     # ANTI-PATTERN: use specific version
    ports:
      - "5432:5432"            # ANTI-PATTERN: exposing DB port to host
    environment:
      POSTGRES_USER: admin
      POSTGRES_PASSWORD: supersecret123
      POSTGRES_DB: acme_prod
    volumes:
      - pgdata:/var/lib/postgresql/data
      # ANTI-PATTERN: mounting host /etc (security risk)
      - /etc/localtime:/etc/localtime:ro
    # ANTI-PATTERN: no health check for database

  redis:
    image: redis:latest
    ports:
      - "6379:6379"            # ANTI-PATTERN: exposing cache port to host
    command: redis-server --requirepass redispass
    # ANTI-PATTERN: no health check, no persistence config, no resource limits

  rabbitmq:
    image: rabbitmq:management
    ports:
      - "5672:5672"
      - "15672:15672"          # management UI exposed
    environment:
      RABBITMQ_DEFAULT_USER: guest
      RABBITMQ_DEFAULT_PASS: guest   # ANTI-PATTERN: default credentials

  # ANTI-PATTERN: nginx with no health check, host network mode
  nginx:
    image: nginx:latest
    network_mode: host         # ANTI-PATTERN: bypasses Docker networking
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf
    restart: unless-stopped

volumes:
  pgdata:
    driver: local

# ANTI-PATTERN: no custom networks defined — all services share default bridge
# Services that don't need to communicate can still reach each other

# Dockerfile.sample — Dockerfile with common anti-patterns
#
# This Dockerfile contains deliberate bad practices for the Docker
# development skill scanner to detect:
#   - Using "latest" tag
#   - Running as root (no USER directive)
#   - No multi-stage build
#   - Installing unnecessary packages
#   - Not cleaning up apt cache
#   - COPY before dependency install (breaks caching)
#   - Using ADD instead of COPY for local files
#   - No HEALTHCHECK
#   - Exposing secrets in ENV
#   - Multiple RUN layers that should be combined

# ANTI-PATTERN: using "latest" tag — not reproducible
FROM node:latest

# ANTI-PATTERN: setting secrets in ENV (visible in image history)
ENV DATABASE_URL=postgres://admin:s3cretPass@db.internal:5432/appdb
ENV API_KEY=sk-prod-a8f3b2c1d4e5f6a7b8c9d0e1f2a3b4c5

# ANTI-PATTERN: installing unnecessary debug/dev packages in production
RUN apt-get update
RUN apt-get install -y vim curl wget netcat telnet htop strace
RUN apt-get install -y python3 python3-pip
# ANTI-PATTERN: not cleaning apt cache — bloats image
# (missing: && rm -rf /var/lib/apt/lists/*)

WORKDIR /app

# ANTI-PATTERN: copying everything before installing dependencies
# This breaks Docker layer caching — any source change re-installs deps
COPY . .

RUN npm install

# ANTI-PATTERN: using ADD for local files (ADD has extra magic for URLs/tar)
ADD ./config/default.json /app/config/default.json
ADD ./scripts/entrypoint.sh /app/scripts/entrypoint.sh

# ANTI-PATTERN: multiple RUN layers that should be combined
RUN chmod +x /app/scripts/entrypoint.sh
RUN mkdir -p /app/logs
RUN mkdir -p /app/tmp
RUN mkdir -p /app/uploads

# ANTI-PATTERN: exposing too many ports without documentation
EXPOSE 3000
EXPOSE 3001
EXPOSE 9229
EXPOSE 8080

# ANTI-PATTERN: no HEALTHCHECK defined

# ANTI-PATTERN: running as root (no USER directive)
# (missing: USER node or USER 1001)

CMD ["node", "server.js"]

#!/usr/bin/env python3
"""
Docker Compose Validator - Validate docker-compose files for correctness and best practices.

Checks service dependencies, port conflicts, volume configurations,
network definitions, and common misconfigurations.

Author: Claude Skills Engineering Team
License: MIT
"""

import argparse
import json
import re
import sys
from dataclasses import dataclass, asdict
from pathlib import Path
from typing import List, Dict, Any, Optional, Set, Tuple


@dataclass
class Finding:
    """A validation finding."""
    severity: str
    category: str
    service: str
    message: str
    recommendation: str


class ComposeParser:
    """Minimal YAML-like parser for docker-compose files (stdlib only)."""

    def __init__(self, content: str):
        self.content = content
        self.lines = content.split("\n")

    def parse(self) -> Dict[str, Any]:
        """Parse compose file into a structured dict."""
        result: Dict[str, Any] = {}
        current_top_key = None
        current_service = None
        current_sub_key = None
        indent_stack: List[Tuple[int, str]] = []

        for line in self.lines:
            stripped = line.strip()
            if not stripped or stripped.startswith("#"):
                continue

            indent = len(line) - len(line.lstrip())
            # Top-level keys
            if indent == 0 and ":" in stripped:
                key = stripped.split(":")[0].strip()
                current_top_key = key
                result[key] = {}
                current_service = None
                current_sub_key = None
                continue

            if current_top_key == "services" and indent == 2 and ":" in stripped:
                svc_name = stripped.split(":")[0].strip()
                current_service = svc_name
                result.setdefault("services", {})[svc_name] = {}
                current_sub_key = None
                continue

            if current_service and current_top_key == "services":
                svc = result["services"][current_service]
                if indent == 4 and ":" in stripped:
                    key = stripped.split(":")[0].strip()
                    value = ":".join(stripped.split(":")[1:]).strip()
                    current_sub_key = key
                    if value:
                        svc[key] = value
                    else:
                        svc.setdefault(key, [])
                elif indent >= 6 and stripped.startswith("- "):
                    item = stripped[2:].strip()
                    if current_sub_key:
                        if not isinstance(svc.get(current_sub_key), list):
                            svc[current_sub_key] = []
                        svc[current_sub_key].append(item)

        return result


class ComposeValidator:
    """Validates docker-compose configurations."""

    def __init__(self, content: str, check_ports: bool = False):
        self.content = content
        self.check_ports_only = check_ports
        self.findings: List[Finding] = []
        parser = ComposeParser(content)
        self.config = parser.parse()
        self.services = self.config.get("services", {})

    def validate(self) -> List[Finding]:
        """Run all validation checks."""
        if self.check_ports_only:
            self._check_port_conflicts()
            return self.findings

        self._check_structure()
        self._check_port_conflicts()
        self._check_dependencies()
        self._check_volumes()
        self._check_security()
        self._check_best_practices()
        return self.findings

    def _check_structure(self):
        """Check basic compose file structure."""
        if not self.services:
            self.findings.append(Finding(
                severity="critical",
                category="structure",
                service="(global)",
                message="No services defined in compose file.",
                recommendation="Add a 'services' section with at least one service.",
            ))
            return

        for name, svc in self.services.items():
            if not isinstance(svc, dict):
                continue
            has_image = "image" in svc
            has_build = "build" in svc
            if not has_image and not has_build:
                self.findings.append(Finding(
                    severity="critical",
                    category="structure",
                    service=name,
                    message=f"Service '{name}' has neither 'image' nor 'build' defined.",
                    recommendation="Add 'image: <name>:<tag>' or 'build: <context>' to the service.",
                ))

    def _check_port_conflicts(self):
        """Detect duplicate host port bindings."""
        port_map: Dict[str, List[str]] = {}

        for name, svc in self.services.items():
            if not isinstance(svc, dict):
                continue
            ports = svc.get("ports", [])
            if not isinstance(ports, list):
                continue
            for port_spec in ports:
                port_str = str(port_spec).strip().strip('"').strip("'")
                # Parse host port from spec like "8080:80" or "127.0.0.1:8080:80"
                parts = port_str.split(":")
                if len(parts) >= 2:
                    host_port = parts[-2]
                    # Handle port ranges
                    host_key = f"0.0.0.0:{host_port}"
                    if len(parts) == 3:
                        host_key = f"{parts[0]}:{parts[1]}"
                    port_map.setdefault(host_key, []).append(name)

        for port, services in port_map.items():
            if len(services) > 1:
                self.findings.append(Finding(
                    severity="critical",
                    category="ports",
                    service=", ".join(services),
                    message=f"Port conflict: host port {port} is used by services: {', '.join(services)}.",
                    recommendation="Assign unique host ports to each service.",
                ))

    def _check_dependencies(self):
        """Validate service dependencies and detect circular deps."""
        dep_graph: Dict[str, Set[str]] = {}
        for name, svc in self.services.items():
            if not isinstance(svc, dict):
                continue
            deps = svc.get("depends_on", [])
            if isinstance(deps, list):
                dep_graph[name] = set(deps)
            elif isinstance(deps, str):
                dep_graph[name] = {deps}
            else:
                dep_graph[name] = set()

        # Check for references to undefined services
        all_services = set(self.services.keys())
        for name, deps in dep_graph.items():
            for dep in deps:
                if dep not in all_services:
                    self.findings.append(Finding(
                        severity="critical",
                        category="dependencies",
                        service=name,
                        message=f"Service '{name}' depends on undefined service '{dep}'.",
                        recommendation=f"Define service '{dep}' or remove the dependency.",
                    ))

        # Detect circular dependencies using DFS
        visited: Set[str] = set()
        path: List[str] = []

        def dfs(node: str) -> bool:
            if node in path:
                cycle = path[path.index(node):] + [node]
                self.findings.append(Finding(
                    severity="critical",
                    category="dependencies",
                    service=node,
                    message=f"Circular dependency detected: {' -> '.join(cycle)}.",
                    recommendation="Break the circular dependency chain.",
                ))
                return True
            if node in visited:
                return False
            visited.add(node)
            path.append(node)
            for dep in dep_graph.get(node, set()):
                if dfs(dep):
                    return True
            path.pop()
            return False

        for name in self.services:
            visited.clear()
            path.clear()
            dfs(name)

    def _check_volumes(self):
        """Check volume configurations."""
        for name, svc in self.services.items():
            if not isinstance(svc, dict):
                continue
            volumes = svc.get("volumes", [])
            if not isinstance(volumes, list):
                continue
            for vol in volumes:
                vol_str = str(vol)
                # Check for mounting Docker socket
                if "/var/run/docker.sock" in vol_str:
                    self.findings.append(Finding(
                        severity="warning",
                        category="security",
                        service=name,
                        message=f"Service '{name}' mounts Docker socket.",
                        recommendation="Docker socket access grants container root-equivalent privileges. Ensure this is necessary.",
                    ))
                # Check for mounting sensitive host paths
                sensitive_paths = ["/etc/shadow", "/etc/passwd", "/root"]
                for sp in sensitive_paths:
                    if vol_str.startswith(sp + ":") or f":{sp}" in vol_str:
                        self.findings.append(Finding(
                            severity="critical",
                            category="security",
                            service=name,
                            message=f"Service '{name}' mounts sensitive host path: {sp}.",
                            recommendation="Avoid mounting sensitive host paths into containers.",
                        ))

    def _check_security(self):
        """Check security-related configurations."""
        for name, svc in self.services.items():
            if not isinstance(svc, dict):
                continue

            # Check for privileged mode
            if svc.get("privileged") in ("true", True):
                self.findings.append(Finding(
                    severity="critical",
                    category="security",
                    service=name,
                    message=f"Service '{name}' runs in privileged mode.",
                    recommendation="Remove privileged mode. Use specific capabilities instead (cap_add).",
                ))

            # Check for host network mode
            if svc.get("network_mode") == "host":
                self.findings.append(Finding(
                    severity="warning",
                    category="security",
                    service=name,
                    message=f"Service '{name}' uses host network mode.",
                    recommendation="Use bridge or custom networks for better isolation.",
                ))

    def _check_best_practices(self):
        """Check general best practices."""
        for name, svc in self.services.items():
            if not isinstance(svc, dict):
                continue

            # Check restart policy
            if "restart" not in svc:
                self.findings.append(Finding(
                    severity="info",
                    category="best-practice",
                    service=name,
                    message=f"Service '{name}' has no restart policy.",
                    recommendation="Add 'restart: unless-stopped' for production services.",
                ))

            # Check for resource limits
            deploy = svc.get("deploy", "")
            if not deploy or "resources" not in str(deploy):
                self.findings.append(Finding(
                    severity="info",
                    category="best-practice",
                    service=name,
                    message=f"Service '{name}' has no resource limits defined.",
                    recommendation="Add deploy.resources.limits for CPU and memory.",
                ))

            # Check image tags
            image = svc.get("image", "")
            if image and ":" not in image:
                self.findings.append(Finding(
                    severity="warning",
                    category="best-practice",
                    service=name,
                    message=f"Service '{name}' uses image '{image}' without version tag.",
                    recommendation="Pin image to a specific version tag.",
                ))


def format_text(findings: List[Finding], services: Dict) -> str:
    """Format results as human-readable text."""
    lines = []
    lines.append("=" * 60)
    lines.append("DOCKER COMPOSE VALIDATION REPORT")
    lines.append("=" * 60)
    lines.append(f"\nServices found: {len(services)}")
    for name in services:
        lines.append(f"  - {name}")

    critical = [f for f in findings if f.severity == "critical"]
    warnings = [f for f in findings if f.severity == "warning"]
    info = [f for f in findings if f.severity == "info"]

    lines.append(f"\nFindings: {len(critical)} critical, {len(warnings)} warnings, {len(info)} info")
    lines.append("-" * 60)

    for severity, group in [("CRITICAL", critical), ("WARNING", warnings), ("INFO", info)]:
        if not group:
            continue
        lines.append(f"\n[{severity}]")
        for f in group:
            lines.append(f"  [{f.category}] {f.service}: {f.message}")
            lines.append(f"    Fix: {f.recommendation}")
            lines.append("")

    if not findings:
        lines.append("\nNo issues found. Compose file follows best practices.")

    lines.append("=" * 60)
    return "\n".join(lines)


def format_json(findings: List[Finding], services: Dict) -> str:
    """Format results as JSON."""
    return json.dumps({
        "services": list(services.keys()),
        "findings": [asdict(f) for f in findings],
        "summary": {
            "total": len(findings),
            "critical": sum(1 for f in findings if f.severity == "critical"),
            "warnings": sum(1 for f in findings if f.severity == "warning"),
            "info": sum(1 for f in findings if f.severity == "info"),
        }
    }, indent=2)


def main():
    parser = argparse.ArgumentParser(
        description="Validate docker-compose files for correctness and best practices."
    )
    parser.add_argument("--file", "-f", required=True, help="Path to docker-compose file")
    parser.add_argument("--format", choices=["text", "json"], default="text", help="Output format")
    parser.add_argument("--check-ports", action="store_true", help="Only check for port conflicts")
    args = parser.parse_args()

    path = Path(args.file)
    if not path.exists():
        print(f"Error: File not found: {args.file}", file=sys.stderr)
        sys.exit(2)

    content = path.read_text()
    validator = ComposeValidator(content, check_ports=args.check_ports)
    findings = validator.validate()

    if args.format == "json":
        print(format_json(findings, validator.services))
    else:
        print(format_text(findings, validator.services))

    if any(f.severity == "critical" for f in findings):
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Dockerfile Analyzer - Analyze Dockerfiles for best practices, security, and optimization.

Scans Dockerfiles for common issues including layer optimization, security
misconfigurations, base image recommendations, and cache efficiency.

Author: Claude Skills Engineering Team
License: MIT
"""

import argparse
import json
import re
import sys
from dataclasses import dataclass, asdict
from pathlib import Path
from typing import List, Optional, Dict, Any


@dataclass
class Finding:
    """A single analysis finding."""
    severity: str  # critical, warning, info
    category: str  # security, optimization, best-practice
    line: int
    instruction: str
    message: str
    recommendation: str


@dataclass
class StageInfo:
    """Information about a build stage."""
    name: Optional[str]
    base_image: str
    line: int
    instruction_count: int
    run_count: int
    copy_count: int


class DockerfileAnalyzer:
    """Analyzes Dockerfiles for best practices and issues."""

    LARGE_BASE_IMAGES = {
        "ubuntu", "debian", "centos", "fedora", "node", "python",
        "ruby", "golang", "java", "openjdk", "php",
    }

    SLIM_ALTERNATIVES = {
        "ubuntu": "ubuntu:22.04 (pin version) or use debian-slim",
        "debian": "debian:bookworm-slim",
        "node": "node:<version>-alpine or node:<version>-slim",
        "python": "python:<version>-slim or python:<version>-alpine",
        "ruby": "ruby:<version>-slim or ruby:<version>-alpine",
        "golang": "golang:<version>-alpine (or multi-stage with scratch)",
        "java": "eclipse-temurin:<version>-jre-alpine",
        "openjdk": "eclipse-temurin:<version>-jre-alpine",
        "php": "php:<version>-alpine",
    }

    SENSITIVE_PATTERNS = [
        (r"(?i)(password|passwd|secret|token|api_key|apikey)\s*=", "Potential secret in ENV or ARG"),
        (r"COPY.*\.(env|pem|key|crt|p12|pfx)", "Sensitive file copied into image"),
        (r"curl.*\|.*sh", "Piping curl to shell is risky"),
        (r"wget.*\|.*sh", "Piping wget to shell is risky"),
    ]

    def __init__(self, content: str, security_only: bool = False):
        self.content = content
        self.lines = content.strip().split("\n")
        self.findings: List[Finding] = []
        self.stages: List[StageInfo] = []
        self.security_only = security_only
        self._parse_stages()

    def _parse_stages(self):
        """Parse multi-stage build information."""
        current_stage = None
        for i, line in enumerate(self.lines, 1):
            stripped = line.strip()
            if stripped.startswith("#") or not stripped:
                continue
            upper = stripped.upper()
            if upper.startswith("FROM "):
                parts = stripped.split()
                image = parts[1] if len(parts) > 1 else "unknown"
                name = None
                if "AS" in [p.upper() for p in parts]:
                    as_idx = next(j for j, p in enumerate(parts) if p.upper() == "AS")
                    if as_idx + 1 < len(parts):
                        name = parts[as_idx + 1]
                if current_stage:
                    self.stages.append(current_stage)
                current_stage = StageInfo(
                    name=name, base_image=image, line=i,
                    instruction_count=0, run_count=0, copy_count=0,
                )
            elif current_stage:
                current_stage.instruction_count += 1
                if upper.startswith("RUN "):
                    current_stage.run_count += 1
                elif upper.startswith("COPY "):
                    current_stage.copy_count += 1
        if current_stage:
            self.stages.append(current_stage)

    def analyze(self) -> List[Finding]:
        """Run all analysis checks."""
        self._check_base_images()
        self._check_security()
        self._check_layer_optimization()
        self._check_cache_efficiency()
        self._check_best_practices()
        return self.findings

    def _check_base_images(self):
        """Check base image selections."""
        for stage in self.stages:
            image = stage.base_image
            tag = ""
            if ":" in image:
                name, tag = image.rsplit(":", 1)
            else:
                name = image
                tag = "latest"

            if tag == "latest" or ":" not in stage.base_image:
                self.findings.append(Finding(
                    severity="warning",
                    category="best-practice",
                    line=stage.line,
                    instruction=f"FROM {stage.base_image}",
                    message="Using 'latest' tag or no tag is non-deterministic.",
                    recommendation="Pin to a specific version tag for reproducible builds.",
                ))

            base_name = name.split("/")[-1]
            if base_name in self.LARGE_BASE_IMAGES and "slim" not in tag and "alpine" not in tag:
                alt = self.SLIM_ALTERNATIVES.get(base_name, "a slim or alpine variant")
                if not self.security_only:
                    self.findings.append(Finding(
                        severity="info",
                        category="optimization",
                        line=stage.line,
                        instruction=f"FROM {stage.base_image}",
                        message=f"Base image '{base_name}' may be larger than necessary.",
                        recommendation=f"Consider using {alt} to reduce image size.",
                    ))

    def _check_security(self):
        """Check for security issues."""
        has_user = False
        has_healthcheck = False

        for i, line in enumerate(self.lines, 1):
            stripped = line.strip()
            if stripped.startswith("#") or not stripped:
                continue
            upper = stripped.upper()

            if upper.startswith("USER ") and not upper.startswith("USER ROOT"):
                has_user = True

            if upper.startswith("HEALTHCHECK "):
                has_healthcheck = True

            # Check for sensitive patterns
            for pattern, msg in self.SENSITIVE_PATTERNS:
                if re.search(pattern, stripped):
                    self.findings.append(Finding(
                        severity="critical",
                        category="security",
                        line=i,
                        instruction=stripped[:80],
                        message=msg,
                        recommendation="Use Docker secrets, build args, or runtime environment variables instead.",
                    ))

            # Check for ADD with URL (prefer COPY or curl)
            if upper.startswith("ADD ") and ("http://" in stripped or "https://" in stripped):
                self.findings.append(Finding(
                    severity="warning",
                    category="security",
                    line=i,
                    instruction=stripped[:80],
                    message="ADD with URL is less transparent than COPY + curl.",
                    recommendation="Use RUN curl/wget to download, then COPY. This provides better caching and verification.",
                ))

            # Check for privileged apt-get
            if "apt-get" in stripped and "--no-install-recommends" not in stripped and "install" in stripped:
                if not self.security_only:
                    self.findings.append(Finding(
                        severity="info",
                        category="optimization",
                        line=i,
                        instruction=stripped[:80],
                        message="apt-get install without --no-install-recommends installs extra packages.",
                        recommendation="Add --no-install-recommends to reduce image size.",
                    ))

        if not has_user:
            self.findings.append(Finding(
                severity="critical",
                category="security",
                line=0,
                instruction="(global)",
                message="No USER instruction found. Container will run as root.",
                recommendation="Add 'RUN addgroup -S app && adduser -S app -G app' and 'USER app' before CMD/ENTRYPOINT.",
            ))

        if not has_healthcheck and not self.security_only:
            self.findings.append(Finding(
                severity="info",
                category="best-practice",
                line=0,
                instruction="(global)",
                message="No HEALTHCHECK instruction found.",
                recommendation="Add HEALTHCHECK to enable container orchestrators to monitor health.",
            ))

    def _check_layer_optimization(self):
        """Check for layer optimization opportunities."""
        if self.security_only:
            return

        consecutive_runs = []
        current_run_streak = 0
        streak_start = 0

        for i, line in enumerate(self.lines, 1):
            stripped = line.strip()
            if stripped.startswith("#") or not stripped:
                continue
            if stripped.upper().startswith("RUN "):
                if current_run_streak == 0:
                    streak_start = i
                current_run_streak += 1
            else:
                if current_run_streak >= 3:
                    consecutive_runs.append((streak_start, current_run_streak))
                current_run_streak = 0

        if current_run_streak >= 3:
            consecutive_runs.append((streak_start, current_run_streak))

        for start, count in consecutive_runs:
            self.findings.append(Finding(
                severity="warning",
                category="optimization",
                line=start,
                instruction=f"{count} consecutive RUN instructions",
                message=f"{count} consecutive RUN instructions create unnecessary layers.",
                recommendation="Combine into a single RUN with && to reduce layer count.",
            ))

    def _check_cache_efficiency(self):
        """Check for cache-busting patterns."""
        if self.security_only:
            return

        copy_all_line = 0
        run_install_after = False

        for i, line in enumerate(self.lines, 1):
            stripped = line.strip()
            if stripped.startswith("#") or not stripped:
                continue
            upper = stripped.upper()

            if upper.startswith("COPY . ") or upper.startswith("COPY ./ "):
                copy_all_line = i

            if copy_all_line and upper.startswith("RUN "):
                if any(cmd in stripped for cmd in ["pip install", "npm install", "yarn install", "go mod download", "bundle install"]):
                    run_install_after = True

        if copy_all_line and run_install_after:
            self.findings.append(Finding(
                severity="warning",
                category="optimization",
                line=copy_all_line,
                instruction="COPY . (followed by dependency install)",
                message="Copying all files before installing dependencies breaks Docker cache.",
                recommendation="Copy dependency files first (requirements.txt, package.json), install, then copy the rest.",
            ))

    def _check_best_practices(self):
        """Check general best practices."""
        if self.security_only:
            return

        has_dockerignore = Path(".dockerignore").exists()
        if not has_dockerignore:
            self.findings.append(Finding(
                severity="info",
                category="best-practice",
                line=0,
                instruction="(project)",
                message="No .dockerignore file found in current directory.",
                recommendation="Create a .dockerignore to exclude .git, node_modules, __pycache__, etc.",
            ))

        if len(self.stages) == 1 and self.stages[0].run_count > 5:
            self.findings.append(Finding(
                severity="info",
                category="optimization",
                line=1,
                instruction="(global)",
                message="Single-stage build with many instructions. Consider multi-stage builds.",
                recommendation="Use a builder stage for compilation and a minimal runtime stage.",
            ))


def format_text(findings: List[Finding], stages: List[StageInfo]) -> str:
    """Format results as human-readable text."""
    lines = []
    lines.append("=" * 60)
    lines.append("DOCKERFILE ANALYSIS REPORT")
    lines.append("=" * 60)

    # Stage summary
    lines.append(f"\nBuild Stages: {len(stages)}")
    for s in stages:
        name = s.name or "(unnamed)"
        lines.append(f"  Stage '{name}': {s.base_image} ({s.instruction_count} instructions)")

    # Findings by severity
    critical = [f for f in findings if f.severity == "critical"]
    warnings = [f for f in findings if f.severity == "warning"]
    info = [f for f in findings if f.severity == "info"]

    lines.append(f"\nFindings: {len(critical)} critical, {len(warnings)} warnings, {len(info)} info")
    lines.append("-" * 60)

    for severity, group in [("CRITICAL", critical), ("WARNING", warnings), ("INFO", info)]:
        if not group:
            continue
        lines.append(f"\n[{severity}]")
        for f in group:
            loc = f"line {f.line}" if f.line > 0 else "global"
            lines.append(f"  [{f.category}] {loc}: {f.message}")
            lines.append(f"    Instruction: {f.instruction}")
            lines.append(f"    Fix: {f.recommendation}")
            lines.append("")

    if not findings:
        lines.append("\nNo issues found. Dockerfile follows best practices.")

    lines.append("=" * 60)
    return "\n".join(lines)


def format_json(findings: List[Finding], stages: List[StageInfo]) -> str:
    """Format results as JSON."""
    return json.dumps({
        "stages": [asdict(s) for s in stages],
        "findings": [asdict(f) for f in findings],
        "summary": {
            "total": len(findings),
            "critical": sum(1 for f in findings if f.severity == "critical"),
            "warnings": sum(1 for f in findings if f.severity == "warning"),
            "info": sum(1 for f in findings if f.severity == "info"),
        }
    }, indent=2)


def main():
    parser = argparse.ArgumentParser(
        description="Analyze Dockerfiles for best practices, security, and optimization."
    )
    parser.add_argument("--file", "-f", required=True, help="Path to Dockerfile")
    parser.add_argument("--format", choices=["text", "json"], default="text", help="Output format")
    parser.add_argument("--security-only", action="store_true", help="Only report security findings")
    args = parser.parse_args()

    path = Path(args.file)
    if not path.exists():
        print(f"Error: File not found: {args.file}", file=sys.stderr)
        sys.exit(1)

    content = path.read_text()
    analyzer = DockerfileAnalyzer(content, security_only=args.security_only)
    findings = analyzer.analyze()

    if args.format == "json":
        print(format_json(findings, analyzer.stages))
    else:
        print(format_text(findings, analyzer.stages))

    # Exit with non-zero if critical findings
    if any(f.severity == "critical" for f in findings):
        sys.exit(1)


if __name__ == "__main__":
    main()