Application Security in SDLC: Shifting Security Left

# secure_sdlc_framework.py
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from enum import Enum
from datetime import datetime
 
class SDLCPhase(Enum):
    REQUIREMENTS = "requirements"
    DESIGN = "design"
    DEVELOPMENT = "development"
    TESTING = "testing"
    DEPLOYMENT = "deployment"
    OPERATIONS = "operations"
 
class SecurityActivity(Enum):
    THREAT_MODELING = "threat_modeling"
    SECURITY_REQUIREMENTS = "security_requirements"
    SECURE_DESIGN_REVIEW = "secure_design_review"
    CODE_REVIEW = "code_review"
    SAST = "static_analysis"
    DAST = "dynamic_analysis"
    PENETRATION_TEST = "penetration_test"
    SECURITY_ASSESSMENT = "security_assessment"
    VULNERABILITY_SCAN = "vulnerability_scan"
    SECURITY_MONITORING = "security_monitoring"
 
@dataclass
class SecurityGate:
    gate_id: str
    phase: SDLCPhase
    required_activities: List[SecurityActivity]
    pass_criteria: Dict
    blocking: bool = True
 
@dataclass
class SecurityFinding:
    finding_id: str
    phase: SDLCPhase
    activity: SecurityActivity
    severity: str
    title: str
    description: str
    remediation: str
    status: str
    assigned_to: Optional[str] = None
    due_date: Optional[datetime] = None
 
class SecureSDLC:
    """Manage security activities across SDLC"""
 
    PHASE_ACTIVITIES = {
        SDLCPhase.REQUIREMENTS: [
            SecurityActivity.SECURITY_REQUIREMENTS,
            SecurityActivity.THREAT_MODELING
        ],
        SDLCPhase.DESIGN: [
            SecurityActivity.THREAT_MODELING,
            SecurityActivity.SECURE_DESIGN_REVIEW
        ],
        SDLCPhase.DEVELOPMENT: [
            SecurityActivity.SAST,
            SecurityActivity.CODE_REVIEW
        ],
        SDLCPhase.TESTING: [
            SecurityActivity.DAST,
            SecurityActivity.PENETRATION_TEST,
            SecurityActivity.VULNERABILITY_SCAN
        ],
        SDLCPhase.DEPLOYMENT: [
            SecurityActivity.SECURITY_ASSESSMENT,
            SecurityActivity.VULNERABILITY_SCAN
        ],
        SDLCPhase.OPERATIONS: [
            SecurityActivity.SECURITY_MONITORING,
            SecurityActivity.VULNERABILITY_SCAN
        ]
    }
 
    def __init__(self):
        self.gates = self._create_default_gates()
        self.findings: List[SecurityFinding] = []
        self.activities_completed = {}
 
    def _create_default_gates(self) -> List[SecurityGate]:
        """Create default security gates"""
        return [
            SecurityGate(
                gate_id="G1",
                phase=SDLCPhase.DESIGN,
                required_activities=[
                    SecurityActivity.THREAT_MODELING,
                    SecurityActivity.SECURITY_REQUIREMENTS
                ],
                pass_criteria={
                    "threat_model_completed": True,
                    "security_requirements_defined": True,
                    "high_risks_addressed": True
                },
                blocking=True
            ),
            SecurityGate(
                gate_id="G2",
                phase=SDLCPhase.DEVELOPMENT,
                required_activities=[
                    SecurityActivity.SAST,
                    SecurityActivity.CODE_REVIEW
                ],
                pass_criteria={
                    "no_critical_findings": True,
                    "no_high_findings": True,
                    "code_review_completed": True
                },
                blocking=True
            ),
            SecurityGate(
                gate_id="G3",
                phase=SDLCPhase.TESTING,
                required_activities=[
                    SecurityActivity.DAST,
                    SecurityActivity.PENETRATION_TEST
                ],
                pass_criteria={
                    "no_critical_findings": True,
                    "high_findings_remediated": True,
                    "pentest_passed": True
                },
                blocking=True
            ),
            SecurityGate(
                gate_id="G4",
                phase=SDLCPhase.DEPLOYMENT,
                required_activities=[
                    SecurityActivity.SECURITY_ASSESSMENT
                ],
                pass_criteria={
                    "all_findings_addressed": True,
                    "security_signoff": True
                },
                blocking=True
            )
        ]
 
    def check_gate(self, phase: SDLCPhase) -> Dict:
        """Check if security gate for phase is passed"""
        gate = next((g for g in self.gates if g.phase == phase), None)
 
        if not gate:
            return {"passed": True, "message": "No gate defined for this phase"}
 
        # Check required activities completed
        completed = self.activities_completed.get(phase, [])
        missing_activities = [
            a for a in gate.required_activities if a not in completed
        ]
 
        if missing_activities:
            return {
                "passed": False,
                "gate_id": gate.gate_id,
                "message": f"Missing activities: {[a.value for a in missing_activities]}",
                "blocking": gate.blocking
            }
 
        # Check findings criteria
        phase_findings = [f for f in self.findings if f.phase == phase]
        critical_findings = [f for f in phase_findings if f.severity == "critical" and f.status != "resolved"]
        high_findings = [f for f in phase_findings if f.severity == "high" and f.status != "resolved"]
 
        if gate.pass_criteria.get("no_critical_findings") and critical_findings:
            return {
                "passed": False,
                "gate_id": gate.gate_id,
                "message": f"{len(critical_findings)} unresolved critical findings",
                "blocking": gate.blocking
            }
 
        if gate.pass_criteria.get("no_high_findings") and high_findings:
            return {
                "passed": False,
                "gate_id": gate.gate_id,
                "message": f"{len(high_findings)} unresolved high findings",
                "blocking": gate.blocking
            }
 
        return {
            "passed": True,
            "gate_id": gate.gate_id,
            "message": "All criteria met"
        }

# threat_modeling.py
from dataclasses import dataclass
from typing import List, Dict
from enum import Enum
 
class STRIDECategory(Enum):
    SPOOFING = "spoofing"
    TAMPERING = "tampering"
    REPUDIATION = "repudiation"
    INFORMATION_DISCLOSURE = "information_disclosure"
    DENIAL_OF_SERVICE = "denial_of_service"
    ELEVATION_OF_PRIVILEGE = "elevation_of_privilege"
 
@dataclass
class DataFlow:
    flow_id: str
    source: str
    destination: str
    data_type: str
    protocol: str
    crosses_trust_boundary: bool
 
@dataclass
class ThreatScenario:
    threat_id: str
    category: STRIDECategory
    affected_component: str
    description: str
    likelihood: str  # low, medium, high
    impact: str      # low, medium, high
    risk_rating: str
    mitigations: List[str]
    status: str
 
class ThreatModeler:
    """STRIDE-based threat modeling"""
 
    STRIDE_QUESTIONS = {
        STRIDECategory.SPOOFING: [
            "Can an attacker pretend to be another user or system?",
            "Is authentication properly implemented?",
            "Can credentials be intercepted or replayed?"
        ],
        STRIDECategory.TAMPERING: [
            "Can data be modified in transit?",
            "Can stored data be altered?",
            "Is input validation sufficient?"
        ],
        STRIDECategory.REPUDIATION: [
            "Can actions be traced to their origin?",
            "Is logging comprehensive and tamper-proof?",
            "Can users deny performing actions?"
        ],
        STRIDECategory.INFORMATION_DISCLOSURE: [
            "Can sensitive data be accessed without authorization?",
            "Is data encrypted at rest and in transit?",
            "Are error messages revealing too much?"
        ],
        STRIDECategory.DENIAL_OF_SERVICE: [
            "Can the system be overwhelmed with requests?",
            "Are there rate limits in place?",
            "Can resources be exhausted?"
        ],
        STRIDECategory.ELEVATION_OF_PRIVILEGE: [
            "Can users gain unauthorized permissions?",
            "Is authorization properly enforced?",
            "Are there privilege escalation paths?"
        ]
    }
 
    def __init__(self, system_name: str):
        self.system_name = system_name
        self.components = []
        self.data_flows = []
        self.threats = []
 
    def add_component(self, component: Dict):
        """Add system component"""
        self.components.append(component)
 
    def add_data_flow(self, flow: DataFlow):
        """Add data flow"""
        self.data_flows.append(flow)
 
    def analyze_threats(self) -> List[ThreatScenario]:
        """Analyze system for STRIDE threats"""
        threats = []
 
        for component in self.components:
            for category in STRIDECategory:
                threat = self._analyze_component_threat(component, category)
                if threat:
                    threats.append(threat)
 
        for flow in self.data_flows:
            if flow.crosses_trust_boundary:
                flow_threats = self._analyze_flow_threats(flow)
                threats.extend(flow_threats)
 
        self.threats = threats
        return threats
 
    def _analyze_component_threat(
        self,
        component: Dict,
        category: STRIDECategory
    ) -> ThreatScenario:
        """Analyze component for specific threat category"""
 
        component_type = component.get("type", "")
        component_name = component.get("name", "")
 
        # Example threat generation logic
        threat_templates = {
            "web_server": {
                STRIDECategory.SPOOFING: "Attacker spoofs legitimate users via session hijacking",
                STRIDECategory.TAMPERING: "Request parameters manipulated to alter behavior",
                STRIDECategory.DENIAL_OF_SERVICE: "Server overwhelmed by flood of requests"
            },
            "database": {
                STRIDECategory.INFORMATION_DISCLOSURE: "SQL injection exposes sensitive data",
                STRIDECategory.TAMPERING: "Unauthorized modification of stored data"
            },
            "api": {
                STRIDECategory.SPOOFING: "API keys compromised or impersonated",
                STRIDECategory.ELEVATION_OF_PRIVILEGE: "Broken access control allows unauthorized actions"
            }
        }
 
        templates = threat_templates.get(component_type, {})
        description = templates.get(category)
 
        if description:
            return ThreatScenario(
                threat_id=f"T-{component_name}-{category.value[:3].upper()}",
                category=category,
                affected_component=component_name,
                description=description,
                likelihood="medium",
                impact="high",
                risk_rating=self._calculate_risk("medium", "high"),
                mitigations=[],
                status="identified"
            )
 
        return None
 
    def _analyze_flow_threats(self, flow: DataFlow) -> List[ThreatScenario]:
        """Analyze data flow crossing trust boundary"""
        threats = []
 
        # Data in transit threats
        if flow.protocol not in ["https", "tls", "ssh"]:
            threats.append(ThreatScenario(
                threat_id=f"T-{flow.flow_id}-TAMPER",
                category=STRIDECategory.TAMPERING,
                affected_component=f"Flow: {flow.source} -> {flow.destination}",
                description=f"Data transmitted without encryption via {flow.protocol}",
                likelihood="high",
                impact="high",
                risk_rating="critical",
                mitigations=["Implement TLS/HTTPS for all data in transit"],
                status="identified"
            ))
 
        return threats
 
    def _calculate_risk(self, likelihood: str, impact: str) -> str:
        """Calculate risk rating from likelihood and impact"""
        matrix = {
            ("low", "low"): "low",
            ("low", "medium"): "low",
            ("low", "high"): "medium",
            ("medium", "low"): "low",
            ("medium", "medium"): "medium",
            ("medium", "high"): "high",
            ("high", "low"): "medium",
            ("high", "medium"): "high",
            ("high", "high"): "critical"
        }
        return matrix.get((likelihood, impact), "medium")
 
    def generate_report(self) -> Dict:
        """Generate threat model report"""
        return {
            "system": self.system_name,
            "generated_at": datetime.utcnow().isoformat(),
            "components": len(self.components),
            "data_flows": len(self.data_flows),
            "total_threats": len(self.threats),
            "threats_by_category": {
                cat.value: len([t for t in self.threats if t.category == cat])
                for cat in STRIDECategory
            },
            "threats_by_risk": {
                risk: len([t for t in self.threats if t.risk_rating == risk])
                for risk in ["critical", "high", "medium", "low"]
            },
            "threats": [
                {
                    "id": t.threat_id,
                    "category": t.category.value,
                    "component": t.affected_component,
                    "description": t.description,
                    "risk": t.risk_rating,
                    "mitigations": t.mitigations
                }
                for t in self.threats
            ]
        }

# secure_coding_checker.py
import re
from dataclasses import dataclass
from typing import List, Dict
 
@dataclass
class CodeViolation:
    rule_id: str
    severity: str
    line_number: int
    description: str
    remediation: str
    code_snippet: str
 
class SecureCodingChecker:
    """Check code against secure coding standards"""
 
    def __init__(self, language: str):
        self.language = language
        self.rules = self._load_rules()
 
    def _load_rules(self) -> List[Dict]:
        """Load language-specific security rules"""
 
        if self.language == "python":
            return [
                {
                    "id": "PY-SEC-001",
                    "severity": "critical",
                    "pattern": r"eval\s*\(",
                    "description": "Use of eval() can execute arbitrary code",
                    "remediation": "Use ast.literal_eval() for safe evaluation or avoid dynamic code execution"
                },
                {
                    "id": "PY-SEC-002",
                    "severity": "critical",
                    "pattern": r"exec\s*\(",
                    "description": "Use of exec() can execute arbitrary code",
                    "remediation": "Avoid dynamic code execution; use safer alternatives"
                },
                {
                    "id": "PY-SEC-003",
                    "severity": "high",
                    "pattern": r"pickle\.loads?\s*\(",
                    "description": "Unsafe deserialization with pickle",
                    "remediation": "Use json for serialization or verify pickle source"
                },
                {
                    "id": "PY-SEC-004",
                    "severity": "high",
                    "pattern": r"subprocess\.(?:call|run|Popen)\s*\([^)]*shell\s*=\s*True",
                    "description": "Shell injection vulnerability",
                    "remediation": "Use shell=False and pass arguments as list"
                },
                {
                    "id": "PY-SEC-005",
                    "severity": "high",
                    "pattern": r"os\.system\s*\(",
                    "description": "Command injection via os.system",
                    "remediation": "Use subprocess with shell=False"
                },
                {
                    "id": "PY-SEC-006",
                    "severity": "medium",
                    "pattern": r"hashlib\.md5\s*\(",
                    "description": "MD5 is cryptographically weak",
                    "remediation": "Use SHA-256 or stronger hash function"
                },
                {
                    "id": "PY-SEC-007",
                    "severity": "critical",
                    "pattern": r"execute\s*\(\s*[\"'].*%s.*[\"']\s*%",
                    "description": "SQL injection via string formatting",
                    "remediation": "Use parameterized queries"
                },
                {
                    "id": "PY-SEC-008",
                    "severity": "high",
                    "pattern": r"verify\s*=\s*False",
                    "description": "SSL verification disabled",
                    "remediation": "Enable SSL certificate verification"
                },
                {
                    "id": "PY-SEC-009",
                    "severity": "medium",
                    "pattern": r"random\.(random|randint|choice)",
                    "description": "Insecure random number generator for security purposes",
                    "remediation": "Use secrets module for security-sensitive randomness"
                },
                {
                    "id": "PY-SEC-010",
                    "severity": "high",
                    "pattern": r"password\s*=\s*[\"'][^\"']+[\"']",
                    "description": "Hardcoded password detected",
                    "remediation": "Use environment variables or secrets management"
                }
            ]
 
        elif self.language == "javascript":
            return [
                {
                    "id": "JS-SEC-001",
                    "severity": "critical",
                    "pattern": r"eval\s*\(",
                    "description": "Use of eval() can execute arbitrary code",
                    "remediation": "Avoid eval(); use JSON.parse() for JSON data"
                },
                {
                    "id": "JS-SEC-002",
                    "severity": "high",
                    "pattern": r"innerHTML\s*=",
                    "description": "Potential XSS via innerHTML",
                    "remediation": "Use textContent or sanitize HTML input"
                },
                {
                    "id": "JS-SEC-003",
                    "severity": "high",
                    "pattern": r"document\.write\s*\(",
                    "description": "document.write can introduce XSS",
                    "remediation": "Use DOM manipulation methods instead"
                },
                {
                    "id": "JS-SEC-004",
                    "severity": "critical",
                    "pattern": r"new\s+Function\s*\(",
                    "description": "Dynamic function creation can execute arbitrary code",
                    "remediation": "Avoid dynamic function creation"
                },
                {
                    "id": "JS-SEC-005",
                    "severity": "medium",
                    "pattern": r"localStorage\.(setItem|getItem)\s*\([^)]*password",
                    "description": "Sensitive data stored in localStorage",
                    "remediation": "Don't store sensitive data in localStorage"
                }
            ]
 
        return []
 
    def check_file(self, content: str, filename: str) -> List[CodeViolation]:
        """Check file content for security violations"""
        violations = []
        lines = content.split('\n')
 
        for rule in self.rules:
            pattern = re.compile(rule["pattern"], re.IGNORECASE)
 
            for line_num, line in enumerate(lines, 1):
                if pattern.search(line):
                    violations.append(CodeViolation(
                        rule_id=rule["id"],
                        severity=rule["severity"],
                        line_number=line_num,
                        description=rule["description"],
                        remediation=rule["remediation"],
                        code_snippet=line.strip()[:100]
                    ))
 
        return violations
 
    def generate_report(self, violations: List[CodeViolation]) -> Dict:
        """Generate security code review report"""
        return {
            "total_violations": len(violations),
            "by_severity": {
                severity: len([v for v in violations if v.severity == severity])
                for severity in ["critical", "high", "medium", "low"]
            },
            "violations": [
                {
                    "rule": v.rule_id,
                    "severity": v.severity,
                    "line": v.line_number,
                    "description": v.description,
                    "fix": v.remediation,
                    "code": v.code_snippet
                }
                for v in violations
            ]
        }

# .github/workflows/security-sdlc.yml
name: Security SDLC Pipeline
 
on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main]
 
jobs:
  # Gate 1: Static Analysis
  sast:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
 
      - name: Run Semgrep SAST
        uses: returntocorp/semgrep-action@v1
        with:
          config: >-
            p/security-audit
            p/secrets
            p/owasp-top-ten
 
      - name: Run Bandit (Python)
        if: hashFiles('**/*.py')
        run: |
          pip install bandit
          bandit -r . -f json -o bandit-results.json || true
 
      - name: Run ESLint Security (JavaScript)
        if: hashFiles('**/*.js')
        run: |
          npm install eslint eslint-plugin-security
          npx eslint --ext .js . --format json -o eslint-results.json || true
 
      - name: Check for hardcoded secrets
        uses: gitleaks/gitleaks-action@v2
 
      - name: Upload SAST results
        uses: actions/upload-artifact@v4
        with:
          name: sast-results
          path: |
            *-results.json
 
  # Gate 2: Dependency Check
  sca:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
 
      - name: Run Snyk SCA
        uses: snyk/actions/node@master
        continue-on-error: true
        env:
          SNYK_TOKEN: ${{ secrets.SNYK_TOKEN }}
        with:
          args: --severity-threshold=high
 
      - name: OWASP Dependency Check
        uses: dependency-check/Dependency-Check_Action@main
        with:
          project: '${{ github.repository }}'
          path: '.'
          format: 'JSON'
 
  # Gate 3: Container Security
  container-scan:
    runs-on: ubuntu-latest
    if: hashFiles('Dockerfile')
    steps:
      - uses: actions/checkout@v4
 
      - name: Build image
        run: docker build -t app:${{ github.sha }} .
 
      - name: Run Trivy container scan
        uses: aquasecurity/trivy-action@master
        with:
          image-ref: 'app:${{ github.sha }}'
          format: 'sarif'
          output: 'trivy-results.sarif'
          severity: 'CRITICAL,HIGH'
 
      - name: Run Hadolint
        uses: hadolint/hadolint-action@v3.1.0
        with:
          dockerfile: Dockerfile
 
  # Security Gate Check
  security-gate:
    needs: [sast, sca, container-scan]
    runs-on: ubuntu-latest
    steps:
      - name: Download all results
        uses: actions/download-artifact@v4
 
      - name: Check security gate
        run: |
          python scripts/check_security_gate.py

# scripts/check_security_gate.py
import json
import sys
from pathlib import Path
 
def check_security_gate():
    """Check if security gate passes"""
 
    critical_count = 0
    high_count = 0
 
    # Check SAST results
    sast_files = Path('sast-results').glob('*.json')
    for file in sast_files:
        with open(file) as f:
            results = json.load(f)
            # Parse based on tool format
            if 'results' in results:
                for finding in results['results']:
                    severity = finding.get('severity', '').lower()
                    if severity == 'critical':
                        critical_count += 1
                    elif severity == 'high':
                        high_count += 1
 
    print(f"\nSecurity Gate Results:")
    print(f"  Critical findings: {critical_count}")
    print(f"  High findings: {high_count}")
 
    # Gate criteria
    if critical_count > 0:
        print("\n❌ GATE FAILED: Critical findings must be resolved")
        return 1
 
    if high_count > 5:
        print(f"\n❌ GATE FAILED: Too many high findings ({high_count} > 5)")
        return 1
 
    print("\n✅ GATE PASSED")
    return 0
 
if __name__ == "__main__":
    sys.exit(check_security_gate())