graph LR
    subgraph Identity
        U[Users] --> IDP[Identity Provider
AAD / Okta / Cognito]
        SVC[Service Accounts] --> IDP
    end

    subgraph Access Control
        IDP --> AUTHZ[Authorization Layer
Unity Catalog / Lake Formation / Ranger]
        AUTHZ --> CLS[Column-Level Security]
        AUTHZ --> RLS[Row-Level Filters]
        AUTHZ --> TBL[Table / Schema ACLs]
    end

    subgraph Storage
        TBL --> ENC[Encrypted Object Store
S3 / ADLS / GCS]
        ENC --> KMS[KMS / Key Vault]
    end

    subgraph Audit
        AUTHZ --> AL[Audit Log
CloudTrail / Diagnostic Logs]
        ENC --> AL
    end

# AWS: IAM-Roles für EC2/EKS statt Access-Keys
# Role an EKS-Service-Account via IRSA attachen
eksctl create iamserviceaccount   --name spark-pipeline   --namespace data-platform   --cluster harbinger-prod   --attach-policy-arn arn:aws:iam::123456789:policy/LakehouseReadWrite   --approve

-- Catalog für Production-Daten erstellen
CREATE CATALOG harbinger_prod;

-- Schema-Level-Access an Data-Engineers
GRANT USE CATALOG ON CATALOG harbinger_prod TO `data-engineers`;
GRANT CREATE SCHEMA ON CATALOG harbinger_prod TO `data-engineers`;

-- Read-Only an Analysten
GRANT USE CATALOG ON CATALOG harbinger_prod TO `analysts`;
GRANT USE SCHEMA ON SCHEMA harbinger_prod.geopolitical TO `analysts`;
GRANT SELECT ON TABLE harbinger_prod.geopolitical.events TO `analysts`;

-- Direct-Storage-Access revoken
REVOKE ALL PRIVILEGES ON EXTERNAL LOCATION raw_s3 FROM `analysts`;

-- E-Mail-Column maskieren für nicht-privilegierte User
CREATE OR REPLACE FUNCTION harbinger_prod.security.mask_email(email STRING)
RETURNS STRING
RETURN CASE
  WHEN IS_MEMBER('pii-readers') THEN email
  ELSE CONCAT(LEFT(email, 2), '****@****.com')
END;

ALTER TABLE harbinger_prod.users.profiles
ALTER COLUMN email SET MASK harbinger_prod.security.mask_email;

-- Row-Filter: Analysten sehen nur Events ihrer assigned Regions
CREATE OR REPLACE FUNCTION harbinger_prod.security.region_filter(region STRING)
RETURNS BOOLEAN
RETURN IS_MEMBER('global-analysts')
  OR EXISTS (
    SELECT 1 FROM harbinger_prod.security.analyst_regions ar
    WHERE ar.user_email = CURRENT_USER()
    AND ar.region = region
  );

ALTER TABLE harbinger_prod.geopolitical.events
ADD ROW FILTER harbinger_prod.security.region_filter ON (region);

# LF-Tags erstellen
aws lakeformation create-lf-tag   --tag-key "Sensitivity"   --tag-values "Public,Internal,Confidential,Restricted"

aws lakeformation create-lf-tag   --tag-key "Domain"   --tag-values "geopolitical,financial,operational"

# Tags zu Resources assignen
aws lakeformation add-lf-tags-to-resource   --resource '{"Table":{"DatabaseName":"harbinger_prod","Name":"classified_events"}}'   --lf-tags '[{"TagKey":"Sensitivity","TagValues":["Restricted"]}]'

# Access via Tags granten
aws lakeformation grant-permissions   --principal '{"DataLakePrincipalIdentifier":"arn:aws:iam::123456789:role/AnalystRole"}'   --resource '{"LFTagPolicy":{"ResourceType":"TABLE","Expression":[{"TagKey":"Sensitivity","TagValues":["Public","Internal"]}]}}'   --permissions SELECT

# Terraform: S3-Bucket mit CMK-Encryption
resource "aws_s3_bucket_server_side_encryption_configuration" "lakehouse" {
  bucket = aws_s3_bucket.lakehouse.id

  rule {
    apply_server_side_encryption_by_default {
      sse_algorithm     = "aws:kms"
      kms_master_key_id = aws_kms_key.lakehouse.arn
    }
    bucket_key_enabled = true  # reduces KMS API calls by ~99%
  }
}

resource "aws_kms_key" "lakehouse" {
  description             = "Harbinger Lakehouse CMK"
  deletion_window_in_days = 30
  enable_key_rotation     = true

  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Sid    = "Enable IAM policies"
        Effect = "Allow"
        Principal = { AWS = "arn:aws:iam::${var.account_id}:root" }
        Action   = "kms:*"
        Resource = "*"
      },
      {
        Sid    = "Deny key deletion by non-admins"
        Effect = "Deny"
        Principal = { AWS = "*" }
        Action   = ["kms:ScheduleKeyDeletion", "kms:DeleteAlias"]
        Resource = "*"
        Condition = {
          StringNotLike = {
            "aws:PrincipalArn" = "arn:aws:iam::${var.account_id}:role/KMSAdmin"
          }
        }
      }
    ]
  })
}

from cryptography.fernet import Fernet
import base64, os

# Key in AWS Secrets Manager, nicht im Code
def encrypt_column(value: str, key: bytes) -> str:
    f = Fernet(base64.urlsafe_b64encode(key))
    return f.encrypt(value.encode()).decode()

# In PySpark
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

encrypt_udf = udf(lambda v: encrypt_column(v, kms_key_bytes), StringType())

df_encrypted = df.withColumn("ssn_encrypted", encrypt_udf("ssn"))                  .drop("ssn")

# Azure: Private-Endpoint für ADLS Gen2
resource "azurerm_private_endpoint" "adls" {
  name                = "harbinger-adls-pe"
  location            = var.location
  resource_group_name = var.resource_group
  subnet_id           = var.private_subnet_id

  private_service_connection {
    name                           = "adls-connection"
    private_connection_resource_id = azurerm_storage_account.lakehouse.id
    subresource_names              = ["dfs"]
    is_manual_connection           = false
  }
}

# Public-Access disablen
resource "azurerm_storage_account_network_rules" "lakehouse" {
  storage_account_id = azurerm_storage_account.lakehouse.id
  default_action     = "Deny"
  bypass             = ["AzureServices"]
  ip_rules           = []
  virtual_network_subnet_ids = [var.private_subnet_id]
}

# Audit-Log-Delivery via Databricks-Account-API enablen
curl -X POST https://accounts.azuredatabricks.net/api/2.0/accounts/${ACCOUNT_ID}/log-delivery   -H "Authorization: Bearer ${TOKEN}"   -d '{
    "log_delivery_configuration": {
      "log_type": "AUDIT_LOGS",
      "output_format": "JSON",
      "delivery_path_prefix": "audit-logs/databricks",
      "storage_configuration_id": "'${STORAGE_CONFIG_ID}'"
    }
  }'

-- Find alle SELECT auf PII-Tables in den letzten 7 Tagen
SELECT
    timestamp,
    userIdentity.email,
    requestParams.commandText,
    sourceIPAddress
FROM harbinger_audit.databricks.audit_events
WHERE timestamp > CURRENT_TIMESTAMP - INTERVAL 7 DAYS
    AND actionName = 'runCommand'
    AND requestParams.commandText ILIKE '%users.profiles%'
ORDER BY timestamp DESC;

-- Anomalous Access: User queryen zu unusual Hours
SELECT
    userIdentity.email,
    HOUR(timestamp) as hour_of_day,
    COUNT(*) as query_count
FROM harbinger_audit.databricks.audit_events
WHERE timestamp > CURRENT_TIMESTAMP - INTERVAL 30 DAYS
    AND actionName = 'runCommand'
GROUP BY 1, 2
HAVING hour_of_day NOT BETWEEN 7 AND 19
ORDER BY query_count DESC;