Generate TF NSGs (#14)

llamazure/tf/BUILD

@@ -0,0 +1,43 @@
+python_sources(
+	name="tf",
+)
+
+python_tests(
+	name="tests",
+)
+
+python_test_utils(
+	name="test_utils",
+)
+
+python_distribution(
+	name="llamazure.tf",
+	repositories=["@llamazure.tf"],
+	dependencies=[":tf"],
+	long_description_path="llamazure/tf/readme.md",
+	provides=python_artifact(
+		name="llamazure.tf",
+		version="0.1.0",
+		description="Generate some azurerm resources without the headache",
+		author="Daniel Goldman",
+		maintainer="Daniel Goldman",
+		classifiers=[
+			"Development Status :: 3 - Alpha",
+			"Programming Language :: Python :: 3.9",
+			"Programming Language :: Python :: 3.10",
+			"Programming Language :: Python :: 3.11",
+			"Programming Language :: Python :: 3.12",
+			"Programming Language :: Python :: 3.13",
+			"Topic :: Utilities",
+			"Topic :: Internet :: Log Analysis",
+		],
+		project_urls={
+			"Homepage": "https://github.com/lilatomic/llamazure",
+			"Repository": "https://github.com/lilatomic/llamazure",
+			"Changelog": "https://github.com/lilatomic/llamazure/blob/main/llamazure/tf/changelog.md",
+			"Issues": "https://github.com/lilatomic/llamazure/issues",
+		},
+		license="Round Robin 2.0.0",
+		long_description_content_type="text/markdown",
+	),
+)

llamazure/tf/changelog.md

@@ -0,0 +1,7 @@
+# 0
+
+## 0.0
+
+### 0.0.1
+
+- feature: easily generate nsg rules

llamazure/tf/conftest.py

@@ -0,0 +1,14 @@
+import os
+
+import pytest
+import yaml
+
+
+@pytest.fixture
+def it_info():
+	"""Fixture: Bundle of config for integration tests"""
+	secrets = os.environ.get("integration_test_secrets")
+	if not secrets:
+		with open("cicd/secrets.yml", mode="r", encoding="utf-8") as f:
+			secrets = f.read()
+	return yaml.safe_load(secrets)

llamazure/tf/models.py

@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from collections import defaultdict
+from dataclasses import dataclass
+
+
+class TFResource(ABC):
+	"""A Terraform resource"""
+
+	name: str
+	t: str
+
+	@abstractmethod
+	def render(self) -> dict:
+		"""Render the resource as JSON-serialisable data"""
+
+	def subresources(self) -> list[TFResource]:
+		"""Child resources"""
+		return []
+
+
+@dataclass
+class AnyTFResource(TFResource):
+	t: str
+	name: str
+	props: dict
+
+	def render(self) -> dict:
+		return self.props
+
+
+@dataclass
+class Terraform:
+	resource: list[TFResource]
+
+	def render(self):
+		"""Render the terraform resources"""
+		rendered_resources = defaultdict(dict)
+
+		def register(resource: TFResource):
+			rendered_resources[resource.t][resource.name] = resource.render()
+			for subresource in resource.subresources():
+				register(subresource)
+
+		for resource in self.resource:
+			register(resource)
+
+		return {
+			"resource": rendered_resources,
+		}
+
+
+def _pluralise(k: str, v: list[str], pluralise: str = "s") -> dict[str, str | list[str]]:
+	"""Format the k-v pair, pluralising the k if necessary"""
+	if len(v) == 1:
+		return {k: v[0]}
+	else:
+		return {k + pluralise: v}

llamazure/tf/models_test.py

@@ -0,0 +1,33 @@
+from llamazure.tf.models import _pluralise
+
+
+class TestPluralise:
+	def test_empty_list(self):
+		# Test case where the list is empty
+		result = _pluralise("apple", [])
+		expected = {"apples": []}  # Pluralised key with empty list
+		assert result == expected
+
+	def test_single_element(self):
+		# Test case where the list has one element
+		result = _pluralise("apple", ["apple"])
+		expected = {"apple": "apple"}
+		assert result == expected
+
+	def test_multiple_elements(self):
+		# Test case where the list has multiple elements
+		result = _pluralise("apple", ["apple", "orange"])
+		expected = {"apples": ["apple", "orange"]}
+		assert result == expected
+
+	def test_single_element_with_es_suffix(self):
+		# Test case where the list has one element and uses the suffix "es"
+		result = _pluralise("box", ["box"], pluralise="es")
+		expected = {"box": "box"}
+		assert result == expected
+
+	def test_multiple_elements_with_es_suffix(self):
+		# Test case where the list has multiple elements and uses the suffix "es"
+		result = _pluralise("box", ["box", "fox"], pluralise="es")
+		expected = {"boxes": ["box", "fox"]}
+		assert result == expected

llamazure/tf/network.py

@@ -0,0 +1,111 @@
+from __future__ import annotations
+
+from collections import defaultdict
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Generic, TypeVar
+
+from llamazure.tf.models import AnyTFResource, TFResource, _pluralise
+
+T = TypeVar("T")
+
+
+class Counter(Generic[T]):
+	"""Incrementing counter, useful for generating priorities"""
+
+	def __init__(self, initial_value=0):
+		self._initial_value = initial_value
+		self._counter: dict[T, int] = defaultdict(lambda: initial_value)
+
+	def incr(self, name: T):
+		"""Get the current value and increment the counter for the given name"""
+		v = self._counter[name]
+		self._counter[name] += 1
+		return v
+
+
+@dataclass
+class NSG(TFResource):
+	"""An azurerm_network_security_group resource"""
+
+	name: str
+	rg: str
+	location: str
+	rules: list[NSGRule]
+	tags: dict[str, str] = field(default_factory=dict)
+
+	@property
+	def t(self) -> str:  # type: ignore[override]
+		return "azurerm_network_security_group"
+
+	def render(self) -> dict:
+		"""Render for tf-json"""
+		return {
+			"name": self.name,
+			"resource_group_name": self.rg,
+			"location": self.location,
+			"security_rule": [],
+			"tags": self.tags,
+		}
+
+	def subresources(self) -> list[TFResource]:
+		counter: Counter[NSGRule.Direction] = Counter(initial_value=100)
+
+		return [
+			AnyTFResource(
+				name="%s-%s" % (self.name, rule.name),
+				t="azurerm_network_security_rule",
+				props=rule.render("${%s.%s.name}" % (self.t, self.name), self.rg, counter.incr(rule.direction)),
+			)
+			for rule in self.rules
+		]
+
+
+@dataclass
+class NSGRule:
+	"""An azurerm_network_security_rule resource"""
+
+	name: str
+	access: Access
+	direction: Direction
+
+	protocol: str = "Tcp"
+	src_ports: list[str] = field(default_factory=lambda: ["*"])
+	src_addrs: list[str] = field(default_factory=lambda: ["*"])
+	src_sgids: list[str] = field(default_factory=lambda: [])
+	dst_ports: list[str] = field(default_factory=lambda: ["*"])
+	dst_addrs: list[str] = field(default_factory=lambda: ["*"])
+	dst_sgids: list[str] = field(default_factory=lambda: [])
+
+	description: str = ""
+
+	class Access(Enum):
+		"""Access type"""
+
+		Allow: str = "Allow"
+		Deny: str = "Deny"
+
+	class Direction(Enum):
+		"""Direction type"""
+
+		Inbound: str = "Inbound"
+		Outbound: str = "Outbound"
+
+	def render(self, nsg_name: str, rg: str, priority: int):
+		"""Render for tf-json"""
+		return {
+			"name": self.name,
+			"description": self.description,
+			"protocol": self.protocol,
+			**_pluralise("source_port_range", self.src_ports),
+			**_pluralise("destination_port_range", self.dst_ports),
+			**_pluralise("source_address_prefix", self.src_addrs, pluralise="es"),
+			**_pluralise("destination_address_prefix", self.dst_addrs, pluralise="es"),
+			"source_application_security_group_ids": self.src_sgids,
+			"destination_application_security_group_ids": self.dst_sgids,
+			"access": self.access.value,
+			"priority": priority,
+			"direction": self.direction.value,
+			"resource_group_name": rg,
+			"network_security_group_name": nsg_name,
+		}

llamazure/tf/network_integration_test.py

@@ -0,0 +1,78 @@
+"""Integration tests for network resources"""
+import json
+import random
+import subprocess
+
+import pytest
+from _pytest.fixtures import fixture
+
+from llamazure.tf.models import Terraform
+from llamazure.tf.network import NSG, NSGRule
+
+
+def test_shim():
+	"""Make pytest succeed even when no tests are selected"""
+
+
+@fixture
+def random_nsg():
+	"""Generate the name of a random NSG"""
+	return "llamazure-tf-{:06d}".format(random.randint(0, int(1e6)))
+
+
+def versions_tf(subscription_id):
+	return (
+		"""\
+terraform {
+	required_providers {
+		azurerm = {
+			source	= "hashicorp/azurerm"
+			version = "=4.0.0"
+		}
+	}
+}
+
+provider "azurerm" {
+	resource_provider_registrations = "none"
+	subscription_id = "%s"
+	features {}
+}
+"""
+		% subscription_id
+	)
+
+
+class TestNetworkIntegration:
+	"""Integration tests for network resources"""
+
+	@pytest.mark.integration
+	def test_network_integration(self, random_nsg, tmp_path, it_info):
+		tf = Terraform(
+			[
+				NSG(
+					random_nsg,
+					"llamazure-tf-test",
+					"Canada Central",
+					rules=[
+						NSGRule(name="Single", access=NSGRule.Access.Allow, direction=NSGRule.Direction.Outbound, src_addrs=["1.1.1.1/32"], src_ports=["443"]),
+						NSGRule(name="Plural", access=NSGRule.Access.Allow, direction=NSGRule.Direction.Outbound, src_addrs=["1.1.1.1/32", "1.1.1.2/32"], src_ports=["80", "443"]),
+					],
+				)
+			]
+		)
+
+		basedir = tmp_path / "tf"
+		basedir.mkdir()
+
+		with open(basedir / "versions.tf", mode="w", encoding="utf-8") as f:
+			f.write(versions_tf(subscription_id=it_info["tf"]["subscription"]))
+
+		with open(basedir / "nsg.tf.json", mode="w", encoding="utf-8") as f:
+			json.dump(tf.render(), f, indent="\t")
+
+		def run_tf(argv: list[str]):
+			subprocess.run(["terraform", f"-chdir={basedir}", *argv], check=True, capture_output=True)
+
+		run_tf(["init"])
+		run_tf(["apply", "-auto-approve"])
+		run_tf(["apply", "-destroy", "-auto-approve"])