کوڈ اور OPA کے بطور پالیسیوں کے ساتھ AI سے تیار کردہ انفراسٹرکچر کا نظم کیسے کریں۔ [Full Handbook]

جدید ترین ماڈل تقریباً 100% نحوی طور پر درست کوڈ تیار کرتے ہیں۔ ویرا کوڈ کی 2026 کی رپورٹ میں واضح طور پر کہا گیا ہے کہ "نحو کو مؤثر طریقے سے حل کیا گیا ہے۔”

یہ ایک سنگ میل کی طرح لگتا ہے، لیکن یہاں یہ کیوں مسئلہ ہے.

اسی رپورٹ میں 100 سے زیادہ ماڈلز کا تجربہ کیا گیا اور پتہ چلا کہ سیکیورٹی پاس کرنے کی اوسط شرح 56% تھی، "پہلی رپورٹ میں 55% سے تھوڑی تبدیلی” اور یہ کہ تقریباً 44% تخلیقی ملازمتوں کے نتیجے میں خطرناک خطرات پیدا ہوئے۔

فنکشنل درستگی اور سیکورٹی الگ الگ مسائل ہیں، جن کے حل ہونے کا امکان صرف ایک ہے۔

نتائج نہ تو باہر ہیں اور نہ ہی نئے۔ 2022 IEEE سیکیورٹی اور رازداری میں، NYU ٹنڈن ٹیم نے GitHub Copilot کو 89 سیکیورٹی سے متعلقہ منظرناموں کے ذریعے چلایا، 1,689 پروگرام بنائے، اور پتہ چلا کہ ان میں سے تقریباً 40% MITER کے CWE ٹاپ 25 میں آئٹمز کے لیے خطرے سے دوچار تھے۔ ACM کمیونیکیشنز تحقیقی جھلکیاں۔

نومبر 2024 میں، جارج ٹاؤن کے سینٹر فار سیکیورٹی اینڈ ایمرجنگ ٹیکنالوجیز نے پانچ LLMs کا جائزہ لیا اور بتایا کہ ان کے تیار کردہ تقریباً نصف ٹکڑوں میں ایسے کیڑے موجود ہیں جو استحصال کا باعث بن سکتے ہیں۔ چار سال، چار آزاد ٹیمیں، چار طریقہ کار، اور ہر بار ایک ہی جواب۔

2023 ACM CCS میں، اسٹینفورڈ کے نیل پیری، میگھا سریواستو، دیپک کمار، اور ڈین بونے نے ایک تجربے میں ڈویلپرز کو شامل کیا۔ 47 شرکاء تھے، 5 سیکورٹی سے متعلق پروگرامنگ ٹاسک کے ساتھ، 3 زبانیں، 33 AI معاونین کے ساتھ، اور 14 بغیر۔ جس گروپ کو مدد ملی اس نے بہت کم محفوظ کوڈ لکھا اور مزید وہ اس بات پر یقین کرنے کا زیادہ امکان رکھتے ہیں کہ وہ جو کوڈ لکھتے ہیں وہ محفوظ ہے۔

یہ ایک چھوٹا مطالعہ ہے اور یہ بتاتا ہے کہ مسئلہ خود حل کیوں نہیں ہو سکتا۔ عام طور پر، وہ طریقہ کار جو اسے پکڑتا ہے (اگر ڈویلپر اس کوڈ کی گہرائی میں دیکھتا ہے جس کے بارے میں وہ فکر مند ہیں) وہی عین طریقہ کار ہے جو ٹول کو بند کر دیتا ہے۔

یہ تمام مطالعات درخواست کوڈ کی پیمائش کرتے ہیں۔ تاہم، انفراسٹرکچر کوڈ ایک زیادہ مشکل معاملہ ہے۔ کیونکہ ایک برا سیکورٹی گروپ کبھی ناکام نہیں ہوتا۔ یہ بالکل تحریری طور پر کام کرتا ہے، جو بھی پوچھتا ہے اسے ٹریفک فراہم کرتا ہے، صرف وہی چیز جو اسے آپ کے خلاف پکڑتی ہے وہ ہے جو آپ کا فرق پڑھتا ہے۔

میں کتاب کا جائزہ نہیں لے سکتا کیونکہ میں نے اسے پڑھ لیا ہے، اور نہ ہی کوئی اور کر سکتا ہے۔ میں کیا کر سکتا ہوں قواعد کو ایک فارمیٹ میں لکھتا ہوں جہاں کمپیوٹر تمام تبدیلیوں کی جانچ کرے گا۔ کوڈ کے طور پر پالیسی کا یہی مطلب ہے۔

یہ ہینڈ بک چیک لکھنے کے عمل میں آپ کی رہنمائی کرتی ہے۔ ہم اسے اصل کمزور ذخیرہ میں تفویض کرتے ہیں، ایک واضح ورژن کی نشاندہی کرتے ہیں جو اس کے سامنے موجود 9 میں سے 5 خلاف ورزیوں کو ٹھیک کرتا ہے، اور پھر اسے ٹھیک کرتا ہے۔

آخر میں، آپ کو معلوم ہو جائے گا کہ کیسے:

  • JSON کو ریگو پالیسیاں لکھیں۔ terraform show -json ہم پیدا کرتے ہیں۔

  • اپنی پالیسی کی جانچ کریں جس طرح آپ فکسچر اور کوریج رپورٹس کا استعمال کرتے ہوئے اپنے ایپلیکیشن کوڈ کی جانچ کرتے ہیں۔

  • CI پائپ لائن قابل اعتماد ایگزٹ کوڈ کنٹریکٹس کا استعمال کرتے ہوئے کمانڈ لائن گیٹس بناتی ہے۔

  • CEL کا استعمال کرتے ہوئے Kubernetes کی منظوری کے دوران غیر تعمیل کام کے بوجھ کو مسدود کریں۔

  • ماڈل کو پالیسی لکھنے دیں۔ opa check اور آپ کی اپنی جانچ اس بات کا تعین کرے گی کہ آپ اسے رکھیں گے یا نہیں۔

  • وہی پالیسی انجن AI ایجنٹس کی ٹول کالز کی منظوری دیتا ہے۔

انڈیکس

شرطیں

آپ کو درج ذیل کی ضرورت ہوگی:

  • ٹرمینلز کے ساتھ کام کرنا python3 (3.10 اور اس سے اوپر)۔

  • jqکمانڈ لائن سے JSON پڑھنے کے لیے۔

  • تقریباً 700MB ڈسک کیونکہ AWS Terraform فراہم کنندہ بڑا ہے۔

  • Anthropic API کلید (صرف 7 مرحلہ)۔ دیگر تمام اقدامات آف لائن چلتے ہیں۔

mkdir policy-lab && cd policy-lab
python3 -m venv .venv
source .venv/bin/activate
pip install anthropic

curl -L -o opa https://openpolicyagent.org/downloads/v1.20.2/opa_darwin_arm64_static
chmod +x opa && sudo mv opa /usr/local/bin/

curl -L -o tf.zip https://releases.hashicorp.com/terraform/1.14.2/terraform_1.14.2_darwin_arm64.zip
unzip tf.zip && sudo mv terraform /usr/local/bin/

ونڈوز پر، آپ ماحول کو استعمال کرتے ہوئے چالو کرتے ہیں: .venvScriptsactivateدونوں ڈاؤن لوڈ یو آر ایل کو اس سے تبدیل کریں: opa_windows_amd64.exe اور terraform_1.14.2_windows_amd64.zip.

میں نے نیچے سب کچھ چلایا۔ OPA 1.20.2، ٹیرافارم 1.14.2، اور AWS فراہم کنندہ 6.xmacOS پر۔ پالیسی نحو OPA 1.x میں مستحکم ہے۔

اگر آپ OPA 0.x استعمال کر رہے ہیں، تو یہاں کے تمام اصول درکار ہیں: import rego.v1 میں اسے سب سے اوپر شامل کروں گا اور اس کے بجائے اپ گریڈ کروں گا۔ خلاف ورزیوں کی تعداد AWS فراہم کنندہ ورژن پر منحصر ہے، جو فراہم کنندہ 5 سے شروع ہوتی ہے اور صرف ایک مستحکم JSON پلان کی شکل میں ہوتی ہے۔

کلیدی اصطلاحات سادہ انگریزی میں

  • کوڈ کے طور پر پالیسی: ایک تنظیم میں پہلے سے طے شدہ قواعد، جو کہ مجوزہ تبدیلیوں کو لاگو کرنے اور فیصلہ واپس کرنے کے لیے ایک پروگرام کے طور پر لکھے گئے ہیں۔

  • لیگو: اوپن پالیسی ایجنٹ کے ذریعہ استفسار کی زبان کا جائزہ لیا گیا۔ یہ اعلانیہ ہے۔ اصول باڈی شرائط کی ایک فہرست ہے جو سب کو پورا کرنا ضروری ہے۔

  • JSON پلان: ٹیرافارم کیا کرے گا اس کی مشین پڑھنے کے قابل تفصیل، جس کے ذریعے تیار کیا گیا ہے: terraform show -json. پالیسی کہتی ہے: .tf فائل نہیں آتی۔

  • داخلہ کنٹرول: Kubernetes API سرور کے اندر ایک نقطہ جہاں کسی چیز کو ذخیرہ کرنے سے پہلے مسترد کیا جا سکتا ہے۔

  • کہ: کامن ایکسپریشن لینگویج، ایک چھوٹی ایکسپریشن لینگوئج جس کا اندازہ Kubernetes مقامی طور پر API سرور کے اندر، بغیر کسی ویب ہکس کے تعینات کرتا ہے۔

  • جھوٹی کسٹم کلیئرنس: وہ وسائل جہاں پالیسی پاس ہوئی لیکن ناکام ہو جانی چاہیے تھی۔ آپ اس کی پیمائش نہیں کر سکتے جب تک کہ آپ اسے تلاش نہ کریں کیونکہ کوئی بھی اسے نوٹس نہیں کرتا ہے۔

ایک ہی فیصلہ تین جگہوں پر کیا گیا ہے، جن میں سے صرف ایک خاص طور پر Kubernetes پر لاگو ہوتا ہے۔

مرحلہ 1: جانچنے کے لیے اصل بنیادی ڈھانچہ حاصل کریں۔

میں کمزور Terraform فائلوں کو تیار نہیں کرنا چاہتا تھا۔ اس کی وجہ یہ ہے کہ فائلوں کو تیار کرنے کا مطلب ہے کیڑے پیدا کرنا، اور پالیسیاں صرف کیڑے پکڑتی ہیں جو آپ لگاتے ہیں۔ لہذا میں نے دوسروں کے لکھے اور شائع کردہ کوڈ کی تلاش کی۔

TerraGoat ایک جان بوجھ کر کمزور Terraform ذخیرہ ہے جسے Bridgecrew نے شائع کیا ہے۔ پن کی ہوئی کمٹ سے EC2 ماڈیول درآمد کریں۔

SHA=729f8da62c6a85ce4af5ad3d123de97776d954c4
curl -s "https://raw.githubusercontent.com/bridgecrewio/terragoat/$SHA/terraform/aws/ec2.tf" 
  | sed -n '77,96p'
resource "aws_security_group" "web-node" {
  # security group is open to the world in SSH port
  name        = "${local.resource_prefix.value}-sg"
  description = "${local.resource_prefix.value} Security Group"
  vpc_id      = aws_vpc.web_vpc.id

  ingress {
    from_port = 80
    to_port   = 80
    protocol  = "tcp"
    cidr_blocks = [
    "0.0.0.0/0"]
  }
  ingress {
    from_port = 22
    to_port   = 22
    protocol  = "tcp"
    cidr_blocks = [
    "0.0.0.0/0"]
  }

دوسری لائن پر تبصرہ TerraGoat کا ہے، جو پورٹ 22 کے دنیا کے لیے کھلے ہونے کی طرف اشارہ کرتا ہے۔

TerraGoat کے ماڈیولز جدید ترین Terraform میں شروع نہیں کیے گئے ہیں۔ کیونکہ یہ اب بھی اعلان کرتا ہے: type = "string" Terraform 0.12 میں فرسودہ اور 1.x میں مسترد کر دیا گیا۔ لہذا میں نے وسائل کو اپنے کم سے کم ماڈیول میں لہرایا، صرف دو حوالوں کو TerraGoat کے اندرونی مقامی لوگوں کی جگہ لے کر۔

بنانا main.tf:

terraform {
  required_version = ">= 1.9"
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 6.0"
    }
  }
}

# Mock credentials. This configuration is only ever planned, never applied,
# so the provider must not try to reach AWS.
provider "aws" {
  region                      = "us-west-2"
  access_key                  = "mock"
  secret_key                  = "mock"
  skip_credentials_validation = true
  skip_metadata_api_check     = true
  skip_requesting_account_id  = true
  skip_region_validation      = true
}

resource "aws_vpc" "web_vpc" {
  cidr_block = "10.0.0.0/16"
}

# Verbatim from bridgecrewio/terragoat, terraform/aws/ec2.tf, commit 729f8da.
# Only the two references to TerraGoat's own locals are replaced with literals.
resource "aws_security_group" "web-node" {
  name        = "terragoat-sg"
  description = "terragoat Security Group"
  vpc_id      = aws_vpc.web_vpc.id

  ingress {
    from_port = 80
    to_port   = 80
    protocol  = "tcp"
    cidr_blocks = [
    "0.0.0.0/0"]
  }
  ingress {
    from_port = 22
    to_port   = 22
    protocol  = "tcp"
    cidr_blocks = [
    "0.0.0.0/0"]
  }
  egress {
    from_port = 0
    to_port   = 0
    protocol  = "-1"
    cidr_blocks = [
    "0.0.0.0/0"]
  }
  depends_on = [aws_vpc.web_vpc]
  tags = {
    git_commit           = "d68d2897add9bc2203a5ed0632a5cdd8ff8cefb0"
    git_file             = "terraform/aws/ec2.tf"
    git_last_modified_at = "2020-06-16 14:46:24"
    git_org              = "bridgecrewio"
    git_repo             = "terragoat"
  }
}

پلان JSON بنائیں۔

terraform init
terraform plan -out=tfplan.binary
terraform show -json tfplan.binary > plan.json

فرضی اسناد اہم ہیں۔ terraform plan چونکہ صرف تخلیق کی ترتیب AWS کو کال نہیں کرتی ہے، skip_credentials_validation اور وہ تین بہن بھائی جن کی فراہم کنندہ تصدیق کرنے کی کوشش نہیں کرتا ہے اور کچھ بھی لاگو نہیں ہوتا ہے۔

مرحلہ 2: پالیسی سے پہلے ٹیسٹ لکھیں۔

میں جو قوانین چاہتا ہوں وہ ہیں: کوئی بھی سیکورٹی گروپ اپنے انتظامی بندرگاہ کو عوامی انٹرنیٹ پر بے نقاب نہیں کر سکتا۔

یہ کوڈ کی ایک لائن کی طرح لگتا ہے، لیکن جانچ سے پتہ چلتا ہے کہ ایسا کیوں نہیں ہوتا ہے۔ بنانا policy/network_test.rego:

package terraform.network_test

import data.terraform.network

plan(resources) := {"resource_changes": resources}

security_group(ingress) := {
	"address": "aws_security_group.web",
	"type": "aws_security_group",
	"change": {"actions": ["create"], "after": {"ingress": [ingress]}},
}

test_denies_ssh_open_to_the_world if {
	fixture := plan([security_group({
		"from_port": 22,
		"to_port": 22,
		"protocol": "tcp",
		"cidr_blocks": ["0.0.0.0/0"],
	})])

	count(network.deny) == 1 with input as fixture
}

# A from_port equality check would miss this. The range check does not.
test_denies_wide_open_port_range if {
	fixture := plan([security_group({
		"from_port": 0,
		"to_port": 65535,
		"protocol": "tcp",
		"cidr_blocks": ["0.0.0.0/0"],
	})])

	count(network.deny) == 4 with input as fixture
}

test_denies_ipv6_route_to_the_world if {
	fixture := plan([security_group({
		"from_port": 22,
		"to_port": 22,
		"protocol": "tcp",
		"ipv6_cidr_blocks": ["::/0"],
	})])

	count(network.deny) == 1 with input as fixture
}

test_denies_standalone_ingress_rule if {
	fixture := plan([{
		"address": "aws_vpc_security_group_ingress_rule.ssh",
		"type": "aws_vpc_security_group_ingress_rule",
		"change": {"actions": ["create"], "after": {
			"from_port": 22,
			"to_port": 22,
			"ip_protocol": "tcp",
			"cidr_ipv4": "0.0.0.0/0",
			"cidr_ipv6": null,
		}},
	}])

	count(network.deny) == 1 with input as fixture
}

test_denies_deprecated_standalone_rule if {
	fixture := plan([{
		"address": "aws_security_group_rule.ssh",
		"type": "aws_security_group_rule",
		"change": {"actions": ["create"], "after": {
			"type": "ingress",
			"from_port": 22,
			"to_port": 22,
			"cidr_blocks": ["0.0.0.0/0"],
		}},
	}])

	count(network.deny) == 1 with input as fixture
}

test_allows_ssh_from_a_private_range if {
	fixture := plan([security_group({
		"from_port": 22,
		"to_port": 22,
		"protocol": "tcp",
		"cidr_blocks": ["10.0.0.0/8"],
	})])

	count(network.deny) == 0 with input as fixture
}

test_allows_https_from_the_world if {
	fixture := plan([security_group({
		"from_port": 443,
		"to_port": 443,
		"protocol": "tcp",
		"cidr_blocks": ["0.0.0.0/0"],
	})])

	count(network.deny) == 0 with input as fixture
}

# An all-protocols rule opens every port, whatever its port fields say.
# The first version of this test asserted the opposite and hid the bug.
test_denies_all_protocols_rule_open_to_the_world if {
	fixture := plan([security_group({
		"from_port": 0,
		"to_port": 0,
		"protocol": "-1",
		"cidr_blocks": ["0.0.0.0/0"],
	})])

	count(network.deny) == 4 with input as fixture
}

# Ports the policy cannot read are reported, never passed.
test_reports_a_world_open_rule_with_unreadable_ports if {
	fixture := plan([security_group({
		"from_port": 22,
		"to_port": null,
		"protocol": "tcp",
		"cidr_blocks": ["0.0.0.0/0"],
	})])

	count(network.deny) == 1 with input as fixture
}

test_reports_string_ports if {
	fixture := plan([security_group({
		"from_port": "22",
		"to_port": "22",
		"protocol": "tcp",
		"cidr_blocks": ["0.0.0.0/0"],
	})])

	count(network.deny) == 1 with input as fixture
}

# Unreadable ports on a rule that is not open to the world stay quiet.
test_ignores_unreadable_ports_on_a_private_range if {
	fixture := plan([security_group({
		"from_port": 22,
		"to_port": null,
		"protocol": "tcp",
		"cidr_blocks": ["10.0.0.0/8"],
	})])

	count(network.deny) == 0 with input as fixture
}

# `considered` drives the gate's pass-or-vacuous decision, so it needs
# tests of its own even though it takes no part in the judgement.
test_considers_every_ingress_bearing_type if {
	fixture := plan([
		security_group({}),
		{"address": "aws_vpc_security_group_ingress_rule.a", "type": "aws_vpc_security_group_ingress_rule", "change": {"actions": ["create"], "after": {}}},
		{"address": "aws_security_group_rule.b", "type": "aws_security_group_rule", "change": {"actions": ["create"], "after": {}}},
	])

	count(network.considered) == 3 with input as fixture
}

test_does_not_consider_unrelated_types if {
	fixture := plan([{
		"address": "aws_vpc.main",
		"type": "aws_vpc",
		"change": {"actions": ["create"], "after": {}},
	}])

	count(network.considered) == 0 with input as fixture
}

یہ ہے وہ فائل جو انکوڈ کرتی ہے:

  1. with input as fixture جعلی پلان کو ایک ہی اظہار سے بدل دیں۔ یہ کلاؤڈ اکاؤنٹ کے بغیر اپنی پالیسی کو جانچنے کا ایک طریقہ ہے۔

  2. test_denies_wide_open_port_range میں اس کا منتظر ہوں۔ ہاں فی منیجمنٹ پورٹ پر ایک خلاف ورزی ہوتی ہے۔ اس کی وجہ یہ ہے کہ داخلے کے قواعد دائرہ کار کی وضاحت کرتے ہیں۔ کوئی راستہ نہیں 0-65535 یہ قاعدہ SSH کو بالکل واضح پورٹ 22 اصول کی طرح کھولتا ہے، مساوات کی جانچ کو پاس کرتے ہوئے۔

  3. تین ٹیسٹ تین چیزوں کا احاطہ کرتے ہیں: مختلف ایک نظر جسے Terraform اسی خیال کے لیے استعمال کرتا ہے: IPv6 فیلڈ، جدید اسٹینڈ۔ aws_vpc_security_group_ingress_ruleاور فرسودہ aws_security_group_rule. میں نے اسے پہلے نہیں لکھا، اور میں وضاحت کرتا ہوں کہ یہ مرحلہ 9 میں کہاں سے آیا ہے۔

  4. دو test_allows_ کیس اتنا ہی اہم ہے جتنا کہ رد کرنا۔ ایسی پالیسی جو ہر چیز سے انکار کرتی ہے بیکار ہے کیونکہ یہ انکار کے تمام امتحانات سے گزرتی ہے۔

  5. آخری تین ٹیسٹ پالیسی کے پہلے ہی "ہونے” کے بعد آئے اور مرحلہ 9 بتاتا ہے کہ وہ کہاں سے آئے ہیں۔ تمام پروٹوکول قواعد کو پورٹ فیلڈ میں متعین تمام بندرگاہوں کو کھولنا چاہئے، اور بندرگاہوں کے ساتھ قواعد جن کو پالیسی نہیں پڑھ سکتی ہے اس کی اطلاع دی جانی چاہئے۔

مرحلہ 3: ٹیسٹ پاس ہونے تک پالیسیاں بنائیں

Terraform چار شکلوں میں داخل ہونے کی وضاحت کرتا ہے، لہذا پالیسی چاروں کو ایک سیٹ میں معمول بناتی ہے اور پھر فیصلہ کرتی ہے کہ ایک بار سیٹ کیا جاتا ہے۔ بنانا policy/network.rego:

# METADATA
# title: No admin port is reachable from the public internet
# description: |
#   Terraform describes ingress in four different shapes. Each one is
#   normalised into a single `exposures` set first, so the judgement below
#   is written once and a new shape only costs one more helper rule.
#
#   Two things here are deliberate rather than incidental. An all-protocols
#   rule covers every port whatever its port fields say, and a rule whose
#   ports this policy cannot read is reported rather than passed.
package terraform.network

admin_ports := {22, 3389, 3306, 5432}

public_cidrs := {"0.0.0.0/0", "::/0"}

# The AWS provider writes from_port 0 and to_port 0 for an all-protocols
# rule, which opens every port, so the port fields cannot be read literally.
all_protocols := {"-1", "all"}

# Shape 1 and 2: inline ingress blocks, IPv4 and IPv6.
exposures contains exposure if {
	some resource in input.resource_changes
	resource.type == "aws_security_group"
	some ingress in resource.change.after.ingress
	some field in ["cidr_blocks", "ipv6_cidr_blocks"]
	some cidr in object.get(ingress, field, [])
	exposure := {
		"address": resource.address,
		"protocol": object.get(ingress, "protocol", ""),
		"from_port": object.get(ingress, "from_port", null),
		"to_port": object.get(ingress, "to_port", null),
		"cidr": cidr,
	}
}

# Shape 3: the standalone rule the AWS provider has recommended since v5.
exposures contains exposure if {
	some resource in input.resource_changes
	resource.type == "aws_vpc_security_group_ingress_rule"
	some field in ["cidr_ipv4", "cidr_ipv6"]
	cidr := object.get(resource.change.after, field, null)
	is_string(cidr)
	exposure := {
		"address": resource.address,
		"protocol": object.get(resource.change.after, "ip_protocol", ""),
		"from_port": object.get(resource.change.after, "from_port", null),
		"to_port": object.get(resource.change.after, "to_port", null),
		"cidr": cidr,
	}
}

# Shape 4: the deprecated standalone rule, still in most existing estates.
exposures contains exposure if {
	some resource in input.resource_changes
	resource.type == "aws_security_group_rule"
	resource.change.after.type == "ingress"
	some cidr in object.get(resource.change.after, "cidr_blocks", [])
	exposure := {
		"address": resource.address,
		"protocol": object.get(resource.change.after, "protocol", ""),
		"from_port": object.get(resource.change.after, "from_port", null),
		"to_port": object.get(resource.change.after, "to_port", null),
		"cidr": cidr,
	}
}

# The ports a rule really covers. Undefined when the policy cannot tell.
covered_ports(exposure) := [0, 65535] if {
	exposure.protocol in all_protocols
}

covered_ports(exposure) := [exposure.from_port, exposure.to_port] if {
	not exposure.protocol in all_protocols
	is_number(exposure.from_port)
	is_number(exposure.to_port)
}

deny contains msg if {
	some exposure in exposures
	exposure.cidr in public_cidrs

	# A rule covers a port if that port falls inside [from_port, to_port].
	range := covered_ports(exposure)
	some port in admin_ports
	port >= range[0]
	port <= range[1]

	msg := sprintf(
		"%s: ingress rule exposes port %d to %s",
		[exposure.address, port, exposure.cidr],
	)
}

# A rule open to the world whose ports this policy cannot read is reported.
# Passing it would be the policy guessing in the permissive direction.
deny contains msg if {
	some exposure in exposures
	exposure.cidr in public_cidrs
	not covered_ports(exposure)

	msg := sprintf(
		"%s: ingress rule to %s has ports this policy cannot evaluate (%v to %v)",
		[exposure.address, exposure.cidr, exposure.from_port, exposure.to_port],
	)
}

# Addresses this policy knows how to inspect. The gate uses this to tell
# "nothing violated" apart from "nothing examined".
considered contains resource.address if {
	some resource in input.resource_changes
	resource.type in {
		"aws_security_group",
		"aws_vpc_security_group_ingress_rule",
		"aws_security_group_rule",
	}
}

اوپر سے پڑھنا:

  1. ریگو کے قواعد کا باڈی ایک کنکشن ہے۔ تمام لائنوں کو برقرار رکھنا ضروری ہے، some ... in چونکہ لائنیں دہرائی جاتی ہیں، او پی اے وسائل کے تمام مجموعوں، داخلی قوانین، فیلڈز اور بندرگاہوں کو تلاش کرتا ہے۔

  2. 3 الگ exposures قواعد ان کے درمیان ایک سیٹ کی وضاحت کرتے ہیں، جسے ریگو ایک اضافی تعریف کہتے ہیں۔ اگر آپ بعد میں پانچویں شکل شامل کرتے ہیں، تو آپ ایک اور بلاک شامل کریں گے اور اس کے نیچے کچھ نہیں بدلے گا۔

  3. object.get(ingress, field, []) اگر فیلڈ غائب ہے، تو یہ ایک خالی فہرست لوٹاتا ہے، اس لیے صرف IPv4 کے اصول اس وقت غلطی نہیں کریں گے جب پالیسی اس کی تلاش کرے گی: ipv6_cidr_blocks.

  4. covered_ports یہ یہاں ایک حفاظتی والو ہے کیونکہ تمام پروٹوکول قوانین اس کی اطلاع دیتے ہیں۔ 0 کو 0 چونکہ تمام بندرگاہیں کھلتے وقت پلان میں شامل ہوتی ہیں، اس لیے پورٹ فیلڈ کو لفظی طور پر نہیں پڑھا جا سکتا، اور ایک اصول جہاں پورٹ null ہو یا سٹرنگ فنکشن کو غیر متعینہ چھوڑ دیتی ہے۔ deny قوانین خلاف ورزیوں میں بدل جاتے ہیں۔

  5. considered یہ فیصلے کا حصہ نہیں ہے۔ یہ پالیسی ریکارڈ کرتی ہے کہ وہ کن وسائل کے بارے میں بات کر سکتی ہے اور وہ کون سے وسائل استعمال کرتی ہے جو مرحلہ 5 میں غیر حاصل شدہ پاسوں کی اطلاع دینے سے بچنے کے لیے استعمال کرتی ہے۔

چلائیں:

opa test policy
opa check --strict policy
opa fmt --diff policy
PASS: 20/20

شامل کریں -v کو opa test ایک لائن فی ٹیسٹ۔

opa check --strict جبکہ یہ غیر محفوظ متغیرات اور پوشیدہ درآمدات کو پکڑتا ہے۔ opa fmt --diff اگر فارمیٹ پہلے سے ہی معیاری ہے تو کچھ بھی پرنٹ نہیں کیا جائے گا۔ OPA فارمیٹ ریگو کو بطور ٹیبز۔ ہر چیز سے پہلے دونوں کا تعلق CI سے ہے۔

دوسری پالیسی ملکیت کی ٹیگنگ ہے، تو policy/tags.rego:

# METADATA
# title: Every managed resource carries ownership tags
# description: |
#   Terraform emits `tags: null` for a resource with no tags at all, so a
#   policy that reaches into `after.tags` skips exactly the resources with
#   the worst tagging. `tags_of` coerces that null to an empty object.
package terraform.tags

required_tags := {"owner", "cost-center", "data-classification"}

# Resource types that genuinely cannot carry tags.
untaggable := {"aws_iam_policy_attachment", "aws_route_table_association"}

in_scope contains resource if {
	some resource in input.resource_changes
	some action in resource.change.actions
	action in {"create", "update"}
	not resource.type in untaggable
}

tags_of(resource) := tags if {
	tags := resource.change.after.tags
	is_object(tags)
} else := {}

deny contains msg if {
	some resource in in_scope
	some tag in required_tags
	value := object.get(tags_of(resource), tag, "")
	trim_space(value) == ""
	msg := sprintf("%s: missing required tag %q", [resource.address, tag])
}

considered contains resource.address if {
	some resource in in_scope
}

یہاں یہ ہے کہ دو لائنیں اس طرح کیوں نظر آتی ہیں:

  1. trim_space(value) == "" چیک کرو۔ ایسا ہی ہونا چاہیے۔ کیونکہ یہ صرف ریگو کے ساتھ ہی ممکن ہے۔ false undefined غلط ہے، لہذا ایک خالی تار سچ ہے۔ بے نقاب وجود کی تصدیق آسانی سے قبول کی جاتی ہے۔ owner = ""یہ تعمیل تھیٹر کی قسم ہے جسے روکنے کے لیے ٹیگ پالیسیاں موجود ہیں۔

  2. tags_ofاس کا else := {} برانچ ایک بگ کے لیے ایک فکس ہے جو میں نے لکھا تھا اور صرف مرحلہ 9 میں پایا گیا تھا۔

مرحلہ 4: اپنے اصل منصوبے کی نشاندہی کریں۔

ٹیرا گوٹ پلان کے خلاف ایک پیکج کا اندازہ لگائیں۔

opa eval --data policy --input plan.json --format pretty 'data.terraform.network.deny'
[
  "aws_security_group.web-node: ingress rule exposes port 22 to 0.0.0.0/0"
]

پالیسی یہ ہے کہ ایک خلاف ورزی کی اطلاع دیں اور پورٹ 80 پر خاموش رہیں۔ کیونکہ یہ ایک عوامی ویب سرور کا دل ہے، اسی وسیلے سے دنیا کے سامنے آتا ہے۔ ایک چیک جو دونوں کو جھنڈا لگاتا ہے ایک ایسا چیک ہے جسے لوگ نظر انداز کرنا سیکھتے ہیں۔

فی پیکیج ایک کمانڈ چلانے سے پیمانہ نہیں ہوتا ہے، اور ریگو ایک ہی سوال میں پوری نام کی جگہ کو جمع کر سکتا ہے۔

opa eval --data policy --input plan.json --format pretty 
  'union({v | v := data.terraform[_].deny})'
[
  "aws_security_group.web-node: ingress rule exposes port 22 to 0.0.0.0/0",
  "aws_security_group.web-node: missing required tag "cost-center"",
  "aws_security_group.web-node: missing required tag "data-classification"",
  "aws_security_group.web-node: missing required tag "owner"",
  "aws_vpc.web_vpc: missing required tag "cost-center"",
  "aws_vpc.web_vpc: missing required tag "data-classification"",
  "aws_vpc.web_vpc: missing required tag "owner""
]

{v | v := data.terraform[_].deny} یہ جمع کرنے اور سمجھنے کی صلاحیت ہے۔ deny ذیل میں تمام پیکجوں میں ترتیبات data.terraformاور union انہیں فلیٹ بناتا ہے۔ اگر آپ ایک نئی پالیسی فائل کو اس نام کی جگہ میں ڈالتے ہیں، تو اسے کمانڈ تبدیل کیے بغیر اٹھایا جائے گا۔

مرحلہ 5: فیصلوں کو ایگزٹ کوڈز میں تبدیل کریں۔

CI گیٹس ایگزٹ سٹیٹس کے ذریعے بات چیت کرتے ہیں، اور دونوں قسم کی ناکامیوں کو ایک کوڈ میں یکجا کرنا یہ ہے کہ ٹوٹی ہوئی پائپ لائن ایک مہینے میں کیسے گزرتی ہے۔ یہ آلہ استعمال کرتا ہے:

پاس ورڈ فیصلہ معنی
0 پاس میں نے پالیسی چلائی اور وسائل کو چیک کیا، لیکن کچھ نہیں ملا۔
1 ناکام ہم نے پالیسی کو نافذ کیا اور ایک خلاف ورزی کا پتہ چلا۔
2 کچھ نہیں کرنا پالیسی چلائی گئی، لیکن کچھ بھی چیک نہیں کیا گیا، لہذا نتائج بے معنی ہیں۔
2 غلط ٹول یا اس کا ان پٹ دستیاب نہیں ہے۔

چوتھی قطار وہ ہے جہاں زیادہ تر دروازے غلط ہو جاتے ہیں۔ اگر پلان کے پاس وہ وسائل نہیں ہیں جن کے بارے میں پالیسی کو معلوم ہے، رپورٹنگ پاس کرنے کے لیے اس یقین دہانی کی ضرورت ہوتی ہے کہ عملدرآمد فراہم نہیں کر سکتا۔ اس کا اپنا فیصلہ نام ہے اور اس میں ختم کرنے والا صفر نہیں ہے۔

بنانا policy_gate.py:

"""Evaluate a Terraform plan against a directory of Rego policies."""

import argparse
import json
import pathlib
import shutil
import subprocess
import sys

PASS, FAIL, BROKEN = 0, 1, 2

DENY_QUERY = "union({v | v := data.terraform[_].deny})"
CONSIDERED_QUERY = "union({v | v := data.terraform[_].considered})"


def die(message: str) -> None:
    print(f"policy-gate: {message}", file=sys.stderr)
    sys.exit(BROKEN)


def load_plan(path: pathlib.Path) -> dict:
    try:
        text = path.read_text()
    except OSError as exc:
        die(f"cannot read {path}: {exc.strerror}")
    try:
        return json.loads(text)
    except json.JSONDecodeError as exc:
        die(f"{path}:{exc.lineno}:{exc.colno}: invalid JSON: {exc.msg}")


def query(opa: str, policy_dirs: list[pathlib.Path], plan: pathlib.Path, expr: str) -> list:
    command = [opa, "eval", "--input", str(plan), "--format", "raw"]
    for directory in policy_dirs:
        command += ["--data", str(directory)]
    command.append(expr)

    result = subprocess.run(command, capture_output=True, text=True)
    if result.returncode != 0:
        die(f"opa failed: {result.stderr.strip() or result.stdout.strip()}")
    values = json.loads(result.stdout)
    # A rule that yields anything but strings is a policy bug, and sorting a
    # mixed list would surface it as an unrelated TypeError.
    for value in values:
        if not isinstance(value, str):
            die(f"{expr} produced a {type(value).__name__}; "
                "deny and considered rules must yield strings")
    return values


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--plan", required=True, type=pathlib.Path,
                        help="JSON from `terraform show -json`")
    parser.add_argument("--policy", required=True, nargs="+", type=pathlib.Path,
                        help="one or more directories of .rego files")
    parser.add_argument("--opa", default="opa", help="path to the opa binary")
    args = parser.parse_args()

    if shutil.which(args.opa) is None:
        die(f"{args.opa} is not on PATH")
    for directory in args.policy:
        if not directory.is_dir():
            die(f"{directory} is not a directory")

    plan = load_plan(args.plan)
    if "resource_changes" not in plan:
        die(f"{args.plan} has no resource_changes key; is it a Terraform plan?")

    considered = query(args.opa, args.policy, args.plan, CONSIDERED_QUERY)
    if not considered:
        # Reporting a pass here would claim an assurance the run cannot give.
        print(f"VACUOUS: no policy examined any of the "
              f"{len(plan['resource_changes'])} planned resource(s)", file=sys.stderr)
        return BROKEN

    violations = sorted(query(args.opa, args.policy, args.plan, DENY_QUERY))
    if violations:
        print(f"FAIL: {len(violations)} violation(s) "
              f"across {len(considered)} examined resource(s)", file=sys.stderr)
        for violation in violations:
            print(f"  - {violation}", file=sys.stderr)
        return FAIL

    print(f"PASS: {len(considered)} resource(s) examined, no violations")
    return PASS


if __name__ == "__main__":
    sys.exit(main())

اسکرپٹ کی خصوصیات میں شامل ہیں:

  1. argparse سکور --plan اور --policy پسند required=Trueاور --policy یہ لیتا ہے nargs="+"لہذا، پالیسی کی خالی فہرست تجزیہ کرتے وقت ایک خرابی کا سبب بنے گی۔

  2. load_plan JSON نحوی غلطیوں کی قطار اور کالم رپورٹ کرتا ہے۔ json.JSONDecodeError لے جانا lineno اور colno ایک گیٹ جو صرف "Bad JSON" کہتا ہے کسی کی دوپہر کو ضائع کرنے والا ہے۔

  3. ناکامی کے تمام راستے اس سے گزرے ہیں: dieیہ ہمیشہ 2 کے ساتھ ختم ہوتا ہے۔ opaناقابل پڑھنے فائلیں اور resource_changes اس کے بنیادی طور پر، یہ سب ٹول میں ہی غلطی ہے۔

  4. کہ considered استفسار پر عمل درآمد پہلے کہ deny سوال اگر کچھ بھی چیک نہیں کیا جاتا ہے، تو پھانسی ختم ہوجاتی ہے۔ VACUOUS آپ کو اپنا پاس پرنٹ کرنے کا موقع بھی نہیں ملتا۔

  5. خلاف ورزیوں کو ترتیب دیا گیا ہے، لہذا ایک ہی منصوبہ ہر رن میں ایک ہی بائٹ وار آؤٹ پٹ پیدا کرے گا، اور دو CI لاگز کے درمیان فرق معنی خیز ہوگا۔

پالیسی لیب کے عنوان سے ایک ٹرمینل ونڈو۔ جب آپ اوپا ٹیسٹ پالیسی چلاتے ہیں، تو یہ PASS کولن 20 سلیش 20 کی اطلاع دیتا ہے۔ TerraGoat پلان کے ساتھ python3policy_gate.py چلانے سے دو چیک شدہ وسائل پر FAIL کالون 7 کی خلاف ورزی پرنٹ ہوتی ہے اور پورٹ 22 سے 0.0.0.0/0 کو ظاہر کرنے والے ایک داخلی اصول کی فہرست دیتا ہے aws_security_group.web-node اور aws_vpc.web_vpc سے۔ ایکو ڈالر کا سوالیہ نشان واپسی 1

اس پلان میں دو وسائل میں 7 خلاف ورزیاں ہیں، ایگزٹ کوڈ کے ساتھ CI کارروائی کر سکتا ہے۔

پالیسی لیب کے عنوان سے ایک ٹرمینل ونڈو۔ python3policy_gate.py کو خالی پلان کے خلاف چلانا 0 منصوبہ بند وسائل کو چیک کیے بغیر VACUOUS کالون نوپالیسی پرنٹ کرتا ہے، اور ایکو ڈالر کا سوالیہ نشان 0 کی بجائے 2 واپس کرتا ہے۔

یہ ایک گیٹ کے ساتھ خالی منصوبے پر وہی ٹول ہے جو "پاس" کا جواب دیتا ہے۔

GitHub ایکشنز ورک فلو پالیسیوں کو استعمال کرتے ہوئے فیصلے کرنے سے پہلے جانچتا ہے۔

name: policy

on: [pull_request]

jobs:
  policy:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v5

      - name: Install OPA
        run: |
          curl -L -o /usr/local/bin/opa 
            https://openpolicyagent.org/downloads/v1.20.2/opa_linux_amd64_static
          chmod +x /usr/local/bin/opa

      # The policies are code. Lint and test them before trusting them.
      - name: Check policy syntax
        run: opa check --strict policy

      - name: Verify formatting
        run: opa fmt --fail --diff policy

      - name: Test policies
        run: opa test policy --verbose --coverage --format json > coverage.json

      # Only now does anything get judged.
      - name: Evaluate Terraform plan
        run: python3 policy_gate.py --plan plan.json --policy policy

اسے بلاک کرنے سے پہلے صرف دو ہفتے کے لیے گیٹ رپورٹنگ کے ساتھ رول آؤٹ کریں۔ جو پالیسیاں واضح طور پر درست معلوم ہوتی ہیں وہ پراپرٹی کے ساختی حصوں پر ناکام ہو جائیں گی، اور آپ اسے بلاک شدہ ریلیز کے بجائے لاگ لائن میں تلاش کرنے سے بہتر ہوں گے۔

مرحلہ 6: داخلے کے وقت لاگو کیا گیا۔

مرحلہ 5 میں گیٹ اس بات کی تصدیق کرتا ہے کہ آپ کیا تقسیم کرنا چاہتے ہیں۔ یہ نظر نہیں آتا kubectl apply کسی کے لیپ ٹاپ سے، کسی وینڈر کے ہیلم چارٹ سے، یا ایک آپریٹر جو اپنے شیڈول کے مطابق پوڈز بناتا ہے۔ اس کے لیے داخلہ کنٹرول کی ضرورت ہے، اور Kubernetes اب اس میں بلٹ ان ہے۔

ValidatingAdmissionPolicy اس کے بعد سے یہ عام طور پر دستیاب ہو گیا ہے۔ v1.30کسی بھی ویب ہکس کے بغیر اپنے API سرور کے اندر CELs کو تعینات کرنے یا فعال رکھنے کا اندازہ کریں۔

apiVersion: admissionregistration.k8s.io/v1
kind: ValidatingAdmissionPolicy
metadata:
  name: require-trusted-registry
spec:
  failurePolicy: Fail
  matchConstraints:
    resourceRules:
      - apiGroups: [""]
        apiVersions: ["v1"]
        operations: ["CREATE", "UPDATE"]
        resources: ["pods"]
  variables:
    # A Pod has three container lists. A policy that reads only
    # spec.containers is bypassed by moving the image to an initContainer.
    - name: allImages
      expression: >-
        object.spec.containers.map(c, c.image) +
        object.spec.?initContainers.orValue([]).map(c, c.image) +
        object.spec.?ephemeralContainers.orValue([]).map(c, c.image)
  validations:
    - expression: >-
        variables.allImages.all(i, i.startsWith('registry.internal.example.com/'))
      messageExpression: >-
        'images must come from registry.internal.example.com: ' +
        variables.allImages.filter(i,
          !i.startsWith('registry.internal.example.com/')).join(', ')
      reason: Forbidden

پالیسی اس وقت تک کچھ نہیں کرتی جب تک کہ پابندی کو چالو نہ کیا جائے۔ یہ آپ کو ایک نام کی جگہ کی جانچ کرنے کی اجازت دیتا ہے۔

apiVersion: admissionregistration.k8s.io/v1
kind: ValidatingAdmissionPolicyBinding
metadata:
  name: require-trusted-registry-binding
spec:
  policyName: require-trusted-registry
  validationActions: ["Deny"]
  matchResources:
    namespaceSelector:
      matchLabels:
        policy.example.com/enforce: "true"

سیٹ validationActions: ["Warn", "Audit"]ایک نام کی جگہ پر لیبل لگائیں، ایک ہفتے تک اس کا مشاہدہ کریں، اور پھر اگلے پر جائیں۔ ["Deny"] سلیکٹر کو پھیلائیں۔

کہ initContainers اس کی وجہ یہ ہے کہ مسئلہ کو ہینڈل کرنا امیج پرووینس پالیسیوں کو نظرانداز کرنے کا سب سے عام طریقہ ہے۔ ? اختیاری فیلڈ نحو .orValue([]) مکمل اظہار کی غلطی کے بغیر ممکنہ غلطیوں کی فہرست کو پڑھنے کا ایک طریقہ یہ ہے۔

ان دونوں مینی فیسٹس کو لاگو نہیں کیا جا سکا کیونکہ پروسیس کرنے کے لیے کوئی کلسٹر نہیں تھے۔ اسے v1 ریفرنس اسکیما کے خلاف چیک کیا گیا تھا اور اصل API سرور نے اس کی اجازت نہیں دی تھی، اس لیے ہم نے اسے ایک نقطہ آغاز سمجھا اور Warn موڈ (کچھ جو آپ کو بہرحال کرنا ہے)۔

Kyverno سے شروع کرتے ہوئے، دو دیگر انجن بڑے پیمانے پر پیداوار میں ہیں۔ کہ مارچ 2026 میں CNCF گریجویشن اسے بلومبرگ، کوائن بیس، ڈوئچے ٹیلی کام، لنکڈ ان، اور اسپاٹائف کے ذریعے پیداواری مقاصد کے لیے استعمال کیا جاتا ہے۔ پالیسیاں YAML میں لکھی جاتی ہیں، اس لیے پلیٹ فارم ٹیم کو کسی نئی زبان کی ضرورت نہیں ہے، اور بلٹ ان پالیسیاں تخلیق، تصویری دستخط کی توثیق، اور صفائی کو ہینڈل کرتی ہیں، اسے تنہا چھوڑ دیتی ہیں۔

جب آپ ایک ریگو کوڈبیس چاہتے ہیں جو کبرنیٹس کا احاطہ کرتا ہے، تو OPA گیٹ کیپر جواب ہوتا ہے۔ اور ٹیرافارم اور CI وہ جگہ ہے جہاں یہ ٹیوٹوریل بنتا ہے۔ اتپریورتن بھی اب بلٹ ان ہیں۔ MutatingAdmissionPolicy پر مستحکم v1.36.

مرحلہ 7: ماڈل کو پالیسیاں لکھنے کی اجازت دیں۔

پالیسیاں بورنگ ہیں، اور ماڈل بورنگ ہونے میں اچھے ہیں۔ تو واضح اقدام یہ ہے کہ ماڈل کو یہ لکھنے دیں۔

ایسے سوالات ہیں جن کی آپ تقریباً ایک منٹ میں پیمائش کر سکتے ہیں، بالکل وہی ہے جو میں اس مرحلے کے آخر میں کرتا ہوں۔ عوامی تربیت کے اعداد و شمار میں ریگو کا ایک اہم حصہ ہے۔ لیگو v0یہ ایک ایسی بولی ہے جس کا تجزیہ کرنا بند ہو گیا تھا جب OPA 1.0 جنوری 2025 میں ریلیز ہوا تھا۔ ایک ماڈل جو سب سے عام پیٹرن تک پہنچتا ہے وہ بولیوں تک پہنچ جائے گا جسے موجودہ تجزیہ کار مسترد کرتا ہے۔

یونیورسٹی آف کلابریا گروپ کے ذریعہ 2025 پری پرنٹ؛ ARPaCCinoہم اپنے Terraform کیس اسٹڈی میں اسی اثر کی اطلاع دیتے ہیں۔ جب ٹول کے بغیر ریگو سے درخواست کی گئی تو Qwen3-30B اور GPT-4o ہر ایک نے 5 میں سے 0 نحوی طور پر درست پالیسیاں تیار کیں۔ OPA دستاویز میں تلاش شامل کرنے سے کچھ نہیں بدلتا۔ ایک لوپ فراہم کریں جو ماڈل چل سکے۔ opa check اگر آپ غلطیاں پڑھیں تو وہ 4/5 اور 5/5 ہیں۔

یہ شمار ایک چھوٹے کیس اسٹڈی سے ہیں، لہذا سمت کو نتائج کے پائیدار حصے کے طور پر سمجھیں۔

فیڈ بیک لوپ کے ذریعے مسئلہ حل ہو گیا۔ لوپس کی کوئی قیمت نہیں ہے۔کیونکہ ہم اسے پہلے ہی بنا چکے ہیں۔ opa check --strict، opa fmtاور opa test.

لہذا ایک الٹ کے ساتھ لوپ بنائیں جو اسے قابل اعتماد بناتا ہے۔ میں ٹیسٹ لکھتا ہوں، ماڈل پالیسیاں لکھتا ہے۔ ٹیسٹ فکسچر JSON بلاب کو پڑھتا ہے اور کہتا ہے "ہاں، اسے مسترد کر دینا چاہیے" 3 سیکنڈ کے اندر، اس کا جائزہ لینے کے لیے مخصوص اور سستا ہو جاتا ہے۔ اس کے نیسٹڈ فہم کے ساتھ، ریگو کو پڑھنے کے لیے کافی محنت درکار ہوتی ہے اور اسے غلط پڑھنا آسان ہے۔ لوگوں کو جگہ دیں جہاں جائزہ لینے کے اخراجات کم ہوں، اور مشینیں چلائیں جہاں آؤٹ پٹ کو میکانکی طور پر چیک کیا جا سکے۔

بنانا policy_forge.py:

"""Generate a Rego policy from a rule in English, and keep it only if the toolchain agrees."""

import argparse
import pathlib
import re
import subprocess
import sys
import tempfile

WRITTEN, REJECTED, BROKEN = 0, 1, 2

SYSTEM = """You write Open Policy Agent policies in Rego v1 (OPA 1.0+).

Rules:
- Use `if` on every rule body and `contains` for multi-value rules.
- Do not emit `import rego.v1`; it is redundant on OPA 1.0+.
- The input is the JSON from `terraform show -json`.
- Return one ```rego block and nothing else."""


def extract_rego(reply: str) -> str:
    blocks = re.findall(r"```regon(.*?)```", reply, re.DOTALL)
    if not blocks:
        raise ValueError("model returned no rego block")
    if len(blocks) > 1:
        raise ValueError(f"model returned {len(blocks)} rego blocks; expected one")
    return blocks[0]


def verify(opa: str, policy: str, tests: pathlib.Path) -> tuple[bool, str]:
    with tempfile.TemporaryDirectory() as tmp:
        bundle = pathlib.Path(tmp)
        # The tests keep their own name; the policy gets one that cannot
        # collide with it, whatever the caller named the test file.
        (bundle / "candidate_policy.rego").write_text(policy)
        (bundle / tests.name).write_text(tests.read_text())

        for command in ([opa, "check", "--strict"], [opa, "test"]):
            result = subprocess.run(command + [str(bundle)], capture_output=True, text=True)
            if result.returncode != 0:
                return False, (result.stdout + result.stderr).strip()
    return True, "opa check and opa test both passed"


def forge(rule: str, tests: pathlib.Path, ask, opa: str, attempts: int) -> str:
    transcript = [{
        "role": "user",
        "content": (
            f"Write a Rego policy for this rule:nn{rule}nn"
            f"It must satisfy these tests:nn```regon{tests.read_text()}```"
        ),
    }]

    for attempt in range(1, attempts + 1):
        reply = ask(transcript)
        policy = extract_rego(reply)
        ok, output = verify(opa, policy, tests)
        headline = next(iter(output.splitlines()), "no output from the toolchain")
        print(f"attempt {attempt}: {'PASS' if ok else 'FAIL'} - {headline}", file=sys.stderr)
        if ok:
            return policy
        transcript += [
            {"role": "assistant", "content": reply},
            {"role": "user", "content": f"The toolchain rejected that:nn{output}nnFix it."},
        ]

    raise RuntimeError(f"no policy survived {attempts} attempts")


def claude(model: str):
    import anthropic

    client = anthropic.Anthropic()

    def ask(transcript: list[dict]) -> str:
        response = client.messages.create(
            model=model,
            max_tokens=16000,
            system=[{
                "type": "text",
                "text": SYSTEM,
                "cache_control": {"type": "ephemeral"},
            }],
            thinking={"type": "adaptive"},
            messages=transcript,
        )
        return "".join(b.text for b in response.content if b.type == "text")

    return ask


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--rule", required=True, help="the policy, in one English sentence")
    parser.add_argument("--tests", required=True, type=pathlib.Path,
                        help="a _test.rego file you wrote by hand")
    parser.add_argument("--out", required=True, type=pathlib.Path,
                        help="where to write the policy, only if it passes")
    parser.add_argument("--model", default="claude-opus-5")
    parser.add_argument("--attempts", type=int, default=4)
    parser.add_argument("--opa", default="opa")
    args = parser.parse_args()

    if args.attempts < 1:
        print("policy-forge: --attempts must be at least 1", file=sys.stderr)
        return BROKEN
    if not args.tests.is_file():
        print(f"policy-forge: {args.tests} does not exist", file=sys.stderr)
        return BROKEN

    try:
        policy = forge(args.rule, args.tests, claude(args.model), args.opa, args.attempts)
    except RuntimeError as exc:
        print(f"policy-forge: {exc}; nothing written", file=sys.stderr)
        return REJECTED
    except ValueError as exc:
        print(f"policy-forge: {exc}", file=sys.stderr)
        return BROKEN

    args.out.write_text(policy)
    print(f"policy-forge: verified policy written to {args.out}", file=sys.stderr)
    return WRITTEN


if __name__ == "__main__":
    sys.exit(main())
export ANTHROPIC_API_KEY=...
python3 policy_forge.py 
  --rule "No security group may expose an administrative port to the public internet." 
  --tests policy/network_test.rego 
  --out policy/generated.rego

لوپ کی ضمانت یہ ہے:

  1. تصدیق ایک ذیلی عمل کے طور پر چلتی ہے۔ ماڈل یہ نہیں پوچھتا کہ کیا پالیسی درست ہے۔ opa check اور opa test فیصلہ اور ایگزٹ کوڈ واحد ثبوت ہیں جو لوپ قبول کرتا ہے۔

  2. ناکامی خام، غیر خلاصہ ٹول آؤٹ پٹ کو لوٹاتی ہے۔ کمپائلر کی غلطیاں اور ٹیسٹ میں ناکامیاں سب سے زیادہ سگنل فیڈ بیک ہیں جو ایک ماڈل کو مل سکتا ہے، یا اسے دوسرے طریقے سے کہیں، وہ مددگار بٹس کو ختم کر دیتے ہیں۔

  3. کچھ بھی ڈسک تک نہیں پہنچتا جب تک کہ یہ گزر نہ جائے۔ forge تصدیق شدہ پالیسیاں واپس کر دی جاتی ہیں یا بڑھا دی جاتی ہیں، اس لیے دوبارہ کوشش کرنے کا بجٹ ختم ہونے کی وجہ سے غیر تصدیق شدہ پالیسیوں کو اسٹور میں محفوظ کرنے کا کوئی راستہ نہیں ہے۔

میں نے لوپ کو چلانے کے لیے اسکرپٹ ماڈل کا استعمال کیا تاکہ میں API کلید کے بغیر نتائج کو دوبارہ پیش کر سکوں۔ تین جوابات ایک حقیقت پسندانہ v0 نحوی پالیسی، ایک حقیقت پسندانہ لیکن غلط v1 پالیسی، اور ایک تین سطحی پالیسی تھے۔

attempt 1: FAIL - 2 errors occurred during loading:
attempt 2: FAIL - policy/network_test.rego:63:
attempt 3: PASS - opa check and opa test both passed

پہلی کوشش ریگو v0 تھی، deny[msg] { ... } یہ ایک ایسا فارم ہے جس نے جنوری 2025 میں OPA 1.0 کے جاری ہونے پر تجزیہ کرنا بند کر دیا تھا۔ اور یہ عوامی تعلیمی ڈیٹا میں بہت زیادہ موجود ہے۔ opa check --strict میں نے اسے امتحان میں آنے سے پہلے ہی مسترد کر دیا۔

کوشش 2 ایک درست ریگو v1 تھی۔ سب سے زیادہ انجینئرز کی طرف سے جائزہ پاس کیا لیکن پھر بھی 8 میں سے 4 سویٹ تک۔ کیونکہ ہم نے موازنہ کیا۔ ingress.from_port کے خلاف admin_ports براہ راست، دائرہ کار کو نظر انداز کرتا ہے، صرف پڑھنے کے لیے cidr_blocks. opa check مجھے کوئی شکایت نہیں تھی کیونکہ کوڈ بالکل ٹھیک منظم تھا۔

ایک اچھی طرح سے بنائی گئی پالیسی اب بھی ایک بری پالیسی ہو سکتی ہے، اور صرف وہی چیز جو جانتی ہے کہ آپ اس لوپ میں کیا چاہتے ہیں وہ ٹیسٹوں کا مجموعہ ہے جو آپ لکھتے ہیں۔

ریکوری لوپ نیرس نہیں ہے۔ اس کی وجہ یہ ہے کہ ہر کوشش غلطی کے پیغام پر مشروط ایک نئی تخلیق ہے۔ چوتھی کوشش تیسری کوشش کی کچھ خصوصیات کھو سکتی ہے، کیونکہ جو پہلے سے کام کرچکا ہے وہ آگے نہیں بڑھے گا۔ اس ڈیزائن میں کچھ بھی اس کا پتہ نہیں لگاتا ہے۔ کیونکہ جانچنے کے لیے صرف وہی ٹیسٹ سوٹ ہے جو آپ نے لکھا ہے۔

ہم دوبارہ کوششوں کی تعداد کو محدود کرتے ہیں، اپنے سوٹ کو بڑھانا جاری رکھتے ہیں، اور بنائی گئی ہر پالیسی کو پل درخواست کے طور پر دیکھتے ہیں جسے ضم کرنے سے پہلے کوئی بھی منظور کرتا ہے۔

مرحلہ 8: ایجنٹ سیلف مینیجمنٹ

AI ایجنٹس اداکار اور کال ٹولز بھی ہوتے ہیں، اس لیے ہر ٹول کال اجازت کا فیصلہ بن جاتی ہے جسے کچھ کرنے کے لیے کرنے کی ضرورت ہوتی ہے۔

صنعت تیزی سے اس پر اکٹھا ہوگئی۔ Amazon Bedrock AgentCore پالیسیاں عام طور پر مارچ 2026 میں سیڈر کے گیٹ ویز پر ایجنٹ ٹول کالز کا جائزہ لینے کے لیے جاری کی گئیں۔ ایک عام اوپن سورس پیٹرن ایک MCP ٹولز گیٹ وے کے سامنے ایک OPA سائڈ کار ہے۔

تحقیق انہی حدود کو مزید مضبوط کر رہی ہے۔ Defects4J کے پیچھے یونیورسٹی آف واشنگٹن گروپ سے 2026 کا پری پرنٹ؛ LLM ایجنٹس کے لیے پالیسی کی تعمیل کی ٹول سے بہتر حلور کی مدد سے تصدیق (ونسٹن، ونسٹن، اور جسٹ) قدرتی زبان کی پالیسیوں کو SMT رکاوٹوں میں مرتب کریں اور غیر موافق کالوں کو روکنے کے لیے Z3 سولور کا استعمال کریں۔

وہ ایک دعوی کا اشتراک کرتے ہیں: سسٹم پرامپٹس سے پالیسیاں نافذ نہیں ہوتی ہیں۔ انفورسمنٹ کال پاتھ میں ایک انٹرسیپٹر ہے جو "نہیں" واپس کر سکتا ہے اور کال کو ہونے سے روک سکتا ہے۔

بنانا agent/authz.rego:

# METADATA
# title: Agent tool-call authorisation
# description: |
#   Evaluated once per tool call, before the tool runs. The decision has
#   three values rather than two, because an agent worth deploying will
#   sometimes need to do something that a human, not the policy, should
#   approve.
package agent.authz

tool_grants := {
	"support": {"search_orders", "read_customer", "issue_refund"},
	"analytics": {"search_orders", "run_query"},
}

write_tools := {"issue_refund", "run_query"}

refund_ceiling_cents := 10000

# An unmapped role, an unknown tool or a malformed input all land here.
default decision := {"effect": "deny", "reasons": ["no matching grant"]}

decision := {"effect": effect_for(reasons), "reasons": reasons} if {
	count(granted) > 0
	reasons := escalations
}

granted contains role if {
	some role in input.agent.roles
	input.tool in object.get(tool_grants, role, set())
}

effect_for(reasons) := "allow" if count(reasons) == 0

effect_for(reasons) := "require_approval" if count(reasons) > 0

escalations contains reason if {
	input.tool in write_tools
	not input.session.human_in_loop
	reason := sprintf("%q writes state and the session is unattended", [input.tool])
}

# A refund with no readable amount cannot be checked against the ceiling,
# so it escalates. Silence here would clear the exact call an attacker
# would craft.
escalations contains reason if {
	input.tool == "issue_refund"
	not positive_amount
	reason := "refund amount is missing, unreadable, or not positive"
}

# A negative amount is a charge wearing a refund's name.
positive_amount if {
	amount := object.get(input, ["arguments", "amount_cents"], null)
	is_number(amount)
	amount > 0
}

escalations contains reason if {
	input.tool == "issue_refund"
	amount := object.get(input, ["arguments", "amount_cents"], null)
	is_number(amount)
	amount > refund_ceiling_cents
	reason := sprintf(
		"refund of %d cents exceeds the %d cent ceiling",
		[amount, refund_ceiling_cents],
	)
}

# Keyword matching is a coarse guard, and it is here to show the shape of an
# argument-level rule. Anything holding real data wants a SQL parser: this
# catches `DROP TABLE` and misses a statement that spells it another way.
destructive_sql := `(?i)b(drop|truncate|delete|alter|grant|revoke)b`

escalations contains reason if {
	input.tool == "run_query"
	regex.match(destructive_sql, object.get(input, ["arguments", "statement"], ""))
	reason := "statement contains a destructive SQL keyword"
}

فیصلے کی شرائط بند ہیں اور ہم انہیں یہاں واضح طور پر بیان کریں گے۔

اثر بھیجنے والا کیا کرتا ہے۔
اجازت دیں آلے کو چلائیں
منظوری کی ضرورت ہے۔ روکیں، لوگوں کو دکھائیں کیوں، صرف منظور ہونے پر ہی چلائیں۔
انکار یہ اسے مسترد کرتا ہے اور منظوری کا کوئی راستہ فراہم نہیں کرتا ہے۔

یہاں یہ ہے کہ پالیسی کو اس طرح کیوں بنایا گیا ہے:

  1. default decision مسترد کریں: غیر پہچانے گئے ٹولز، وہ کردار جنہیں آپ نقشہ بنانا بھول گئے، یا خراب ان پٹ سب کچھ موجود ہے۔ ایک پالیسی جو بنیادی طور پر ان پٹ کے لیے فیل پر کھلنے کی اجازت دیتی ہے جس کی کسی کو توقع نہیں ہوتی، یعنی ان پٹ جس کا انتخاب حملہ آور کرتا ہے۔

  2. تین قدریں: بائنری اتھارٹی مفید کارروائیوں کو مسدود کرنے اور خطرناک کارروائیوں کی اجازت دینے کے درمیان انتخاب پر مجبور کرتی ہے، اور اس کی تیسری قدر اعلیٰ خود مختاری والے ایجنٹوں کو جائز بنانا ہے۔

  3. سیٹ پر واپسی کی وجوہات: تمام قابل اطلاق وجوہات کو جمع کیا جاتا ہے. اگر کسی کو صبح 3 بجے منظوری کا پیغام ملتا ہے، تو وہ ایک پیغام دیکھیں گے کہ "250,000 سینٹ کی رقم کی واپسی 10,000 سینٹ کی حد سے تجاوز کر گئی ہے" یہ بتائے گا کہ انہیں کیا کرنا ہے۔ "پالیسی کی خلاف ورزی" نہیں ہے۔

  4. دلائل بھی چیک کیے جاتے ہیں۔ issue_refund یہ دیکھتے ہوئے کہ ایجنٹ حصول کا انتخاب کرتا ہے، کیونکہ یہ £5 پر غیر معمولی اور £2,500 پر سنجیدہ ہے، ایجنٹ کے لیے ٹول کا نام گرانولریٹی بہت موٹا ہے۔

15 ٹیسٹوں میں فیصلے کی میزیں شامل ہیں، بشمول خالی رول لسٹ والے ایجنٹ، صفر کی رقم کے ساتھ رقم کی واپسی، اور سوالات کو چھپانا۔ DROP ایک نئی لائن کے بعد:

opa test agent -v
PASS: 15/15

خدمت کریں اور کال کریں۔

opa run --server --addr localhost:8181 agent/
curl -s localhost:8181/v1/data/agent/authz/decision 
  -d '{"input":{"agent":{"roles":["support"]},"tool":"issue_refund",
       "arguments":{"amount_cents":250000},"session":{"human_in_loop":true}}}' | jq .result
{
  "effect": "require_approval",
  "reasons": [
    "refund of 250000 cents exceeds the 10000 cent ceiling"
  ]
}

پالیسی اس وقت تک غیر فعال ہے جب تک کہ آپ جواب کے ساتھ آگے بڑھنے سے انکار نہ کر دیں۔ یہ کلائنٹ ہے:

import json
import urllib.request

OPA_URL = "http://localhost:8181/v1/data/agent/authz/decision"


class PolicyDenied(Exception):
    pass


class ApprovalRequired(Exception):
    pass


def authorize(agent, tool, arguments, session):
    payload = json.dumps({"input": {
        "agent": agent, "tool": tool,
        "arguments": arguments, "session": session,
    }}).encode()
    req = urllib.request.Request(
        OPA_URL, data=payload, headers={"Content-Type": "application/json"}
    )
    with urllib.request.urlopen(req, timeout=2) as resp:
        body = json.load(resp)

    # OPA returns {} with a 200 when a query matches nothing. Fail closed.
    decision = body.get("result", {"effect": "deny", "reasons": ["policy unavailable"]})

    if decision["effect"] == "deny":
        raise PolicyDenied("; ".join(decision["reasons"]))
    if decision["effect"] == "require_approval":
        raise ApprovalRequired("; ".join(decision["reasons"]))
    return decision
search_orders    -> ALLOWED
issue_refund     -> NEEDS APPROVAL (refund of 250000 cents exceeds the 10000 cent ceiling)
delete_account   -> DENIED (no matching grant)

میمو body.get("result", ...): او پی اے کی واپسی۔ {} جب استفسار کسی چیز سے میل نہیں کھاتا ہے تو اسٹیٹس 200 ہے۔ body["result"] تاثر KeyErrorاور اس بات پر منحصر ہے کہ ایجنٹ کا فریم ورک ان استثنیٰ کو کیسے ہینڈل کرتا ہے جو ناکام ہو سکتے ہیں: کھلا. تمام پرتیں، بشمول تجزیہ، بطور ڈیفالٹ مسترد کر دی جاتی ہیں۔

کال authorize() فریم ورک کے ٹول ایگزیکیوشن ہک میں ٹول فنکشن کے عمل سے پہلے۔ تقریباً 15 لائنوں میں، یہ سسٹم پرامپٹ کی شائستہ تجاویز کو اصل حدود میں بدل دیتا ہے۔

مرحلہ 9: میں نے کیا غلط کیا۔

مرحلہ 2 اور 3 مکمل شدہ پالیسی کو ظاہر کرتے ہیں۔ میں سب سے پہلے آسان (جس ورژن پر زیادہ تر ٹیوٹوریل رک جاتا ہے) تک پہنچ گیا، لیکن اس اور جو آپ ابھی پڑھتے ہیں اس کے درمیان جو فاصلہ ہے وہ یہاں سب سے زیادہ مفید ہے۔

ٹیگنگ کی پالیسی بدترین وسائل کو چھوڑ رہی ہے۔

میرے معصوم in_scope قواعد اس کے ساتھ ختم ہوئے: resource.change.after.tagsاسے "صرف ٹیگ کردہ وسائل" کے طور پر پڑھا جاتا ہے۔

Terraform اصل میں کیا کرتا ہے اس سے بھی بدتر ہے۔ tags: null پر مشتمل وسائل کے لیے: کوئی ٹیگ بالکل نہیں۔ایک غیر متعینہ تلاش رول باڈی کے ناکام ہونے کا سبب بنتی ہے، لہذا وسائل مکمل طور پر دائرہ کار سے باہر ہے۔

ٹیرا گوٹ پلان کے دو وسائل ہیں: ایک سیکورٹی گروپ کے پاس پانچ وسائل ہیں: git_* کوئی ٹیگ نہیں ہیں، کوئی ملکیتی ٹیگ نہیں ہیں، لیکن VPC کچھ بھی پاس نہیں کرتا ہے۔

jq -r '.resource_changes[] | "(.address): tags=(.change.after.tags | type)"' plan.json
aws_security_group.web-node: tags=object
aws_vpc.web_vpc: tags=null
opa eval --data naive  --input plan.json --format pretty 'count(data.terraform.tags.deny)'
opa eval --data policy --input plan.json --format pretty 'count(data.terraform.tags.deny)'
3
6

چونکہ تین لاپتہ تمام مکمل طور پر غیر ٹیگ شدہ وسائل پر تھے، پالیسی نے وسائل کو کچھ ٹیگز کے ساتھ جھنڈا لگایا اور خود بخود غیر ٹیگ شدہ وسائل کو صاف کردیا۔

tags_of اس کے ساتھ else := {} برانچ ایک ترمیم ہے اور تین لائنیں لمبی ہے۔

نیٹ ورک کی پالیسیاں چار میں سے ایک شکل اختیار کرتی ہیں:

میری بولی نیٹ ورک پالیسی پڑھیں aws_security_group اور cidr_blocksیہ وہی ہے جو تمام ٹیوٹوریلز دکھاتے ہیں، لیکن Terraform کے پاس اسی استقبالیہ اصول کو ظاہر کرنے کے چار طریقے ہیں۔

![Diagram titled "Four ways Terraform describes one ingress rule". Four boxes are shown. Top left, aws_security_group.ingress[].cidr_blocks، ہلکے نیلے رنگ سے بھرے ہوئے اور 'پڑھیں' کا لیبل لگا ہوا ہے۔ باقی تینوں کا خاکہ سرخ رنگ میں سرخ ہیچنگ کے ساتھ اور بغیر پڑھے ہوئے لیبل لگایا گیا ہے۔ aws_security_group.ingress[].ipv6_cidr_blocks، aws_vpc_security_group_ingress_rule.cidr_ipv4، اور فرسودہ aws_security_group_rule۔ نیلے رنگ کے بھرنے کا مطلب ہے کہ پالیسی یہاں ظاہر ہوتی ہے اور سرخ رنگ کے ہیچ کا مطلب ہے کہ پالیسی یہاں ظاہر نہیں ہوتی ہے۔](https://cdn.hashnode.com/uploads/covers/5f3a74bfc4d5973f55c91c8c/ab61d36d-78bf-45f7-a073-e6p131er="al7073e")

1 اصول، 4 انکوڈنگز، پہلے سے طے شدہ پالیسی صرف اوپر بائیں اندراج کو پڑھتی ہے۔

میں نے تین سیکیورٹی گروپس کے ساتھ ایک دوسری اصل ترتیب کی منصوبہ بندی کی جو پالیسی کے ناقابل پڑھے ہوئے فارم کا استعمال کرتے ہوئے SSH کو دنیا کے لیے کھولے گی۔ یہ دونوں ورژن میں ہے۔ code/تو یہ دوبارہ پیدا کرتا ہے:

opa eval --data naive --input plan-evasion.json 
  --format pretty 'data.terraform.network.deny'
[]

یہ 3 عوامی طور پر کنیکٹ ایبل SSH پورٹس ہیں، کوئی خلاف ورزی نہیں، پرنٹ شدہ گیٹ۔ PASS اور یہ 0 پر نکلا۔

ایک کھلا سوراخ رکھنے کے لیے ٹیسٹ کریں۔

پورٹ رینجز کے ساتھ دوسرا مسئلہ تھا، اور میرا ٹیسٹ سویٹ ان کا احاطہ کر رہا تھا۔ میں نے مندرجہ ذیل کیس لکھا: test_tolerates_null_portsیہ مندرجہ ذیل استقبالیہ اصولوں پر زور دیتا ہے: protocol: "-1" کوئی خلاف ورزی نہیں ہوئی کیونکہ null پورٹ کے ساتھ موازنہ کرنے سے پالیسی نہیں ٹوٹنی چاہیے۔

پروٹوکول کے تمام اصول تمام بندرگاہوں کو کھولتے ہیں اور AWS فراہم کنندہ انہیں مندرجہ ذیل طور پر لاگ کرتا ہے: from_port: 0 اور to_port: 0. رینج چیک لفظی طور پر ایک واحد پورٹ 0 کے طور پر پڑھتا ہے، جو AWS پر سب سے زیادہ قابل اجازت اصول کو کلین سکور دیتا ہے۔

منصوبہ ہے code/plan-all-protocols.jsonیہ ایک وسیلہ ہے۔

jq -c '.resource_changes[] | select(.type=="aws_security_group")
       | .change.after.ingress[0] | {protocol,from_port,to_port,cidr_blocks}' 
   plan-all-protocols.json
opa eval --data naive --input plan-all-protocols.json 
   --format pretty 'data.terraform.network.deny'
{"protocol":"-1","from_port":0,"to_port":0,"cidr_blocks":["0.0.0.0/0"]}
[]

تمام پروٹوکول اور تمام بندرگاہیں پورے انٹرنیٹ کے لیے کھلی ہیں، اور میں نے جان بوجھ کر جو ٹیسٹ لکھے ہیں، وہ ٹھیک ثابت ہوئے۔ یہاں اصلاحات ہیں: covered_portsپروٹوکول کے تمام اصول پورے دائرہ کار میں نقشہ بناتے ہیں اور ناقابل پڑھے جانے والی بندرگاہوں کے لیے غیر متعینہ ہیں۔ deny غیر متعینہ مقدمات کی اطلاع دینے کا اصول۔ وہ ٹیسٹ ختم ہو گیا، اس کی جگہ چار ٹیسٹ ہو گئے۔

opa eval --data policy --input plan-all-protocols.json --format pretty 'data.terraform.network.deny'
[
  "aws_security_group.wide_open: ingress rule exposes port 22 to 0.0.0.0/0",
  "aws_security_group.wide_open: ingress rule exposes port 3306 to 0.0.0.0/0",
  "aws_security_group.wide_open: ingress rule exposes port 3389 to 0.0.0.0/0",
  "aws_security_group.wide_open: ingress rule exposes port 5432 to 0.0.0.0/0"
]

مرحلہ 7 دعوی کرتا ہے کہ ٹیسٹ سویٹ لوپ میں واحد نمونہ ہے جو جانتا ہے کہ صارف کیا چاہتا ہے۔ یہ اس پراپرٹی کی قیمت ہے۔ ایک خراب ٹیسٹ بھی ایک خراب تصریح ہے، اور اس ذیلی حصے پر کوئی بحث نہیں ہے۔

تعداد واقعی کیسی تھی؟

عنوان افقی بار چارٹ۔

دونوں ورژنوں نے اس پلان میں 1.000 اسکور کیے جس کے لیے میں نے پالیسی بنائی تھی۔ خلا صرف ان منصوبوں میں ظاہر ہوتا ہے جو میں نے نہیں بنائے تھے۔

دو حقیقی منصوبوں میں بے ہودہ پالیسی کو 9 میں سے 4 خلاف ورزیاں ملیں۔. انہوں نے TerraGoat سیکورٹی گروپ میں 1.000 کا سکور حاصل کیا، جو لکھتے وقت میرے ذہن میں تھا، اور دو صورتوں میں جو میں نے نہیں کیا، انہیں 0.000 اور 0.500 کا سکور ملا۔

وہ حصہ جس نے مجھے واقعی حیران کردیا۔

میں نے سوچا کہ ٹیسٹ کوریج نے اسے پکڑ لیا ہوگا، لیکن ایسا نہیں ہوا۔ میں نے اصل میں لکھے گئے دو ٹیسٹوں کا استعمال کرتے ہوئے بولی ٹیگنگ پالیسی کو دوبارہ تشکیل دیا۔

opa test . --coverage --format json | jq '{overall: .coverage}'
{
  "overall": 100
}

سادہ پالیسی نے 100% کوریج ریکارڈ کی، 2 میں سے 2 ٹیسٹ پاس کیے، اور بغیر کسی ٹیگ کے وسائل کو ضائع کر دیا۔

کوریج پیمائش کرتا ہے کہ پالیسی کی کون سی لائن ٹیسٹ پر عمل درآمد کرتی ہے، اور آپ کو غیر متوقع ان پٹ شکلوں کے بارے میں کچھ نہیں بتاتی ہے۔ پالیسی کوڈز کے لیے، یہ مکمل ناکامی کا موڈ ہے۔ کوریج رپورٹنگ کے قابل ہے، اور اس بات کا ثبوت ہے کہ پالیسی اصل میں کام کرتی ہے اسے بنیادی ڈھانچے پر چلانے سے حاصل ہوتی ہے جس کے بارے میں آپ نے نہیں لکھا۔

چیک کی حد

کیونکہ جھوٹی اجازتیں کیچز سے زیادہ اہم ہیں، اس لیے یہ گیٹ اب بھی نہیں کر سکتا:

1. نامعلوم قدریں نظر نہیں آتیں۔

Terraform کسی بھی ایسی چیز کو نشان زد کرتا ہے جسے لاگو وقت کے ذریعے نامعلوم کے طور پر حل نہیں کیا جا سکتا۔ null پلان JSON سے after_unknown نقشہ پالیسی پڑھیں change.after.some_field اگر فیلڈ نامعلوم ہے تو یہ نہیں چلے گا۔

کراس ریسورس رولز کے ساتھ یہ سب سے بڑا مسئلہ ہے، جہاں "تمام بالٹیوں میں عوامی رسائی کے بلاکس ہوتے ہیں" منصوبہ بندی کے وقت واقعی مشکل ہے، کیونکہ بلاکس عام طور پر استعمال ہونے والی بالٹی آئی ڈیز کا حوالہ دیتے ہیں۔ (known after apply).

2. صرف پلان کو دیکھیں۔ میں صرف منصوبہ دیکھ رہا ہوں۔

کوئی بھی چیز جو پائپ لائن کے باہر لاگو ہوتی ہے، کنسول میں تبدیل ہوتی ہے، یا تخلیق کے پوشیدہ ہونے کے بعد سے چلی جاتی ہے۔ منصوبہ بندی کے وقت کی تصدیق اور داخلے کے وقت کی تصدیق ہے۔ مختلف آپریشن کو وقفے وقفے سے نابینا جگہ اور اس حالت کا معائنہ کرنے کے لیے ختم کیا جاتا ہے جس میں دونوں رکھے گئے ہیں۔

3. کنٹرول کی مدت اندرونی طور پر ماپا نہیں جا سکتا.

100 گرین چیک قواعد کے بارے میں کچھ نہیں کہتے ہیں کسی نے نہیں لکھا ہے۔ واضح طور پر کنٹرول کی ضروریات اور پالیسی فائلوں کے درمیان نقشہ جات کو برقرار رکھیں اور شیڈول کے مطابق ان کا آڈٹ کریں۔ کیونکہ گیٹ نہیں جان سکتا کہ کیا مانگا نہیں جاتا۔

4. CIDR فہرست بالکل مماثل ہے۔

public_cidrs قبضہ 0.0.0.0/0 اور ::/0 اس کے علاوہ اور کچھ نہیں۔ قواعد شروع ہوتے ہیں۔ 0.0.0.0/1 انٹرنیٹ تک پہنچیں اور آدھے راستے سے گزریں۔ اس میں توسیع کا مطلب یہ ہے کہ یہ فیصلہ کرنا کہ کون سے سابقہ ​​کی لمبائی کو عوامی سمجھا جاتا ہے اور RFC 1918 اسپیس کو تقسیم کرنا، اور یہ فیصلہ تنظیم کا ہے۔

5. میں نے جو پایا وہ چار شکلیں ہیں۔

کہ exposures سیٹ میں وہ چار انکوڈنگز شامل ہیں جن کی میں تلاش کر رہا تھا، اور AWS اور بھی بہت کچھ پیش کرتا ہے۔ سیکورٹی گروپ کا حوالہ (security_groups)، سابقہ ​​فہرست، اور self ایک قاعدہ کسی بندرگاہ تک پہنچنے کا کوئی بھی طریقہ ہے جسے یہ پالیسی ماڈل نہیں کرتی ہے اور کبھی نہیں دیکھے گی۔ میں پانچویں شکل دیکھنے کی توقع کرتا ہوں جب یہ پہلی بار بڑی جائیدادوں کے خلاف چلائی جائے گی۔

6. ریگولیٹری تاریخوں کو تبدیل کرنا

اگر ہم EU AI قانون کی طرف بڑھ رہے ہیں تو، جولائی 2026 میں اعلان کردہ ڈیجیٹل Omnibus 2 اگست 2026 سے Annex III ہائی رسک ذمہ داریاں متعارف کرائے گا۔ 2 دسمبر 2027اور ضمیمہ I کی ذمہ داریاں 2 اگست 2028 تک لاگو ہوتی ہیں، جبکہ آرٹیکل 50 شفافیت کی ذمہ داری 2 اگست 2026 کی اپنی اصل تاریخ کو برقرار رکھتی ہے۔

اپنے کنٹرولز کو انکوڈ کریں اور جب بھی آپ کو ان کی ضرورت ہو سرکاری ٹائم لائن میں تاریخیں دیکھیں۔ اس میں ایسے معاملات بھی شامل ہیں جہاں پچھلی سہ ماہیوں کی بلاگ پوسٹس (بشمول یہ) دوسری صورت میں بیان کرتی ہیں۔

نتیجہ

یہ ماڈل انفراسٹرکچر کوڈ تیار کرتا ہے جو تقریباً ہر بار پارس ہوتا ہے، تقریباً 56% محفوظ ہوتا ہے، اور کسی سے بھی زیادہ تیزی سے پہنچتا ہے۔ دستی جائزہ نے خلا میں کہیں حقیقی کنٹرول کھو دیا۔ قوانین کا مقصد ہمیشہ قابل عمل ہونا تھا اور حجم ختم ہونے سے مسائل پیدا ہوئے۔

اس ٹیوٹوریل میں آپ کریں گے:

  • کمٹ پر، اصل کمزور سیکورٹی گروپ TerraGoat سے درآمد کیے گئے تھے۔ 729f8da ٹیرافارم 1.14.2 کے لیے منصوبہ بنایا گیا ہے۔

  • 20 پالیسی ٹیسٹ لکھ کر OPA 1.20.2 پاس کیا جس میں ایک داخلی اصول کے لیے چار Terraform انکوڈنگز شامل ہیں۔

  • ہمیں TerraGoat پلان میں سات حقیقی خلاف ورزیاں ملیں، پورٹ 80 کو درست طریقے سے نظر انداز کرتے ہوئے۔

  • جب میں نے کچھ تحقیق نہیں کی تو میں نے 4-فیصلے والے معاہدے کے ساتھ ایک گیٹ بنایا جو 2 بار باہر چلا گیا۔

  • ہم نے 100% ٹیسٹ کوریج کی اطلاع دیتے ہوئے سادہ ورژن کو 9 میں سے 4 خلاف ورزیوں کو دیکھا، اور پھر اسے بڑھا کر 9 میں سے 9 کو تلاش کیا۔

  • میں نے تخلیق لوپ کو جوڑ دیا ہے۔ opa check اور ڈسک پر کیا ختم ہوتا ہے اس کا تعین کرنے کے لیے اپنی جانچ کا استعمال کریں۔

  • وہی انجن مجاز ایجنٹ ٹول کو کال کرتا ہے، جس میں رد کرنے کے لیے 15 ٹیسٹ اور ڈیفالٹس ہوتے ہیں۔

میری پالیسی نے ان کیسوں کو سنبھالا جن کے بارے میں میں نے لکھا تھا، ایک جوڑے کو یاد کیا جس کا میں نے کبھی تصور بھی نہیں کیا تھا، اور مجھے وہ تمام نشانات دیے جو اس نے مجھے دیئے تھے (ٹیسٹ پاس ہوا، 100% کوریج، اور صاف opa check) میں نے اتفاق کیا کہ یہ ٹھیک ہے۔ ایک چیز جس سے میں نے اختلاف کیا وہ انفراسٹرکچر تھا جو کسی اور نے لکھا تھا۔ اپنی پالیسی کو ابتدائی طور پر اس کوڈ پر بیان کریں جسے آپ نے نہیں لکھا اور اسے ناکام ہونے سے بچائیں۔

اسکرپٹ جو تمام کوڈ، پالیسیاں، پلان JSON اور تصاویر لکھتی ہے۔ code/ اس ہینڈ بک کے آگے ڈائریکٹری۔ اعداد و شمار کو مندرجہ ذیل کے طور پر دوبارہ پیش کیا جاتا ہے: python3 build/make_images.py اور python3 build/make_terminals.py. ٹرمینل اسکرین شاٹس سچائی سے زیادہ نہیں ہوسکتے ہیں کیونکہ وہ تعمیراتی وقت پر کمانڈز کو دوبارہ چلاتے ہیں۔

اوپر تک سکرول کریں۔