運用 Testinfra 來確保網路連通性測試

起源是因為做 Azure 網路架構設計，網路線拉完 (ExpressRoute、S2S VPN、VNet Peering 等) 之後，下一個需求多半都是要把 Firewall 開起來，希望所有的連線都要經過 Azure Firewall 確保連線安全性，但基於 Azure Networking 本身的特性和使用者的對於網路連通 (Network Connectivity) 的需求，身為一名記憶力不算特好的架構師，為了確保使用者講的跟實際上運作的結果是一樣的，這時候就要把早年在 Edgecore Network 時期講過的 NetDevOps: Next-Generation Network Engineer 拿來用用了

你應該要知道的小知識

Testinfra

Testinfra 是知名 Python 測試開發框架 pytest 衍生的一個測試引擎，主要對標的功能是另一個以 Ruby 為基礎的 Serverspec，但主要也不是要你會寫程式語言才能用，反正用順手的就好，邏輯上都差不多

雖說 Testinfra 預設是使用 pytest 作為基礎，但也可以跟另一個很常見的 Python 測試框架 unittest 融和在一起，我是以這個寫法為主，如果你覺得不太順眼可以看官方的 examples 自己修改，附上本文會使用到的 Code Snippet pichuang/testinfra-code-snippet

個人常用到的 4 個斷言 (Assertion) 有以下幾種，但不限於，可參考unittest - assert methods:

Assert Methods	表達式	解釋	白話文
assertEqual(a, b)	a == b	可以拿來檢查 HTTP Status Code 為 200	這個網頁很正常
assertNotEqual(a, b)	a != b	可以拿來檢查 HTTP Status Code 不為 200	這個網頁不存在是正常的
assertTrue(x)	bool(x) is True	可以檢查 Boolen	這個服務我要連得到
assertFalse(x)	bool(x) is False	可以檢查 Boolen	這個服務我不應該連的到

實際上執行方式要記得加 -v 才能看到百分比

pytest -v test*.py

此外 testinfra 還支援多種 connection backend，包含: kubectl、openshift、ssh、ansible、docker、podman 等，和多種 Modules 可以檢查，包含: file、group、host、package、process、service、user 等，可以參考官方文件，有需求的可以自行試試

Azure Firewall

Azure Firewall 是 Azure 提供的 PaaS 所提供的企業級 Firewall 能力，可以透過 Azure Resource Manager 將服務帶起來，應對不同的 SKU 可以提供不同層級的保護能力，需要 Azure Firewall Policy 才能正常作動，沒設定半條 Policy 的狀況下是預設 DenyAll

如果要在 Azure Firewall 看到具體的 Log 狀態，你需要另外設定 Log Analytics Workspace，並且在 Diagnostic Setting 設定好 Log Analytics Workspace，才能在 Azure Firewall 看到 Log

Azure Firewall Policy

Azure Firewall 實際上需要搭配 Azure Firewall Policy 針對以下 Rule Types 進行設定:

Rule Types	Default Priority	Protocols	Description
DNAT	100	TCP, UDP	可以透過 DNAT 將 Inbound Traffic 轉換 Azure 內部的 Private IP
Network	200	Any, TCP, UDP, ICMP	可以基於 L3/L4 IP, Port, Protocols 進行流量篩選
Application	300	HTTP, HTTPS	可以基於 FQDN, URL, HTTP, HTTPS 進行流量篩選

實務上規則應該會越寫越多，為了確保是不是真的有符合這些規則，生命有限，可以用 testinfra 從使用者視角來反覆確保這些規則是正確的

Unit Test 案例分析

因為懶得從 Azure Firewall 畫面截圖，故以 Terraform azurerm_firewall_policy_rule_collection_group 來表示 Azure Firewall Policy 的規則寫法，和其對應的測試

使用者 A 說: 底下所有服務都要可以互相 Ping 到

因為 Azure Firewall 預設是 DenyAll，所以這邊要先開放 ICMP 的規則，讓所有的流量都可以互相 Ping 到

network-icmp.tf

resource "azurerm_firewall_policy_rule_collection_group" "fprcg-for-transit" {
  network_rule_collection {
    name     = "network-rule-collection-for-transit"
    priority = 200
    action   = "Allow"
    rule {
      name                  = "network-rule-icmp-any-to-any"
      protocols             = ["ICMP"]
      source_addresses      = ["*"]
      destination_addresses = ["*"]
      destination_ports     = ["*"]
    }
  }
}

以 Testinfra 撰寫 Unittest 就會像下列這樣，其中 testinfra.get_host("local://") 是表示在本機執行測試，self.host.addr("10.73.30.164").is_reachable 等價於 ping -W 1 -c 10.73.30.164

testinfra-icmp.py

class TestNetworkRules(unittest.TestCase):
    def setUp(self):
        self.host = testinfra.get_host("local://")
    def test_to_hub_vm(self):
        self.assertTrue(self.host.addr("10.73.30.164").is_reachable)
    def test_to_spoke1_vm(self):
        self.assertTrue(self.host.addr("10.73.31.4").is_reachable)
    def test_to_spoke2_vm(self):
        self.assertTrue(self.host.addr("10.73.33.4").is_reachable)

使用者 B 說: 我要能連到 Google DNS 8.8.8.8 進行 DNS 解析

這個就是一個常見的規則需求，需要能夠連到特定 IP 的某個 Port 做點事，所以這邊要挑選 Network Rule Collection，開放 8.8.8.8 的 53 Port

network-dns.tf

resource "azurerm_firewall_policy_rule_collection_group" "fprcg-for-transit" {
  network_rule_collection {
    name     = "network-rule-collection-for-google-dns"
    priority = 201
    action   = "Allow"
    rule {
      name                  = "network-rule-icmp-any-to-any"
      protocols             = ["TCP", "UDP"]
      source_addresses      = ["*"]
      destination_addresses = ["8.8.8.8"]
      destination_ports     = ["53"]
    }
  }
}

google_dns.port(53).is_reachable 若有指定 Port 則會使用 nc -w 1 -z <IP> <Port> 來做驗證，而不是 ping。這邊也可以觀察到 8.8.8.8，其實沒有開 80 出出來，只有 53 和 443 以及 ICMP

testinfra-google-dns.py

class TestNetworkRules(unittest.TestCase):
    def setUp(self):
        self.host = testinfra.get_host("local://")
    def test_to_google_dns(self):
        google_dns = self.host.addr("8.8.8.8")
        self.assertTrue(google_dns.is_reachable) # Equal to "ping -W 1 -c 1 8.8.8.8"
        self.assertTrue(google_dns.port(53).is_reachable) # Equal to "nc -w 1 -z 8.8.8.8 53"
        self.assertFalse(google_dns.port(80).is_reachable) # Equal to "nc -w 1 -z 8.8.8.8 80"
        self.assertTrue(google_dns.port(443).is_reachable) # Equal to "nc -w 1 -z 8.8.8.8 443"

使用者 C 說: 我要能連到 github.com 進行 git clone 的操作

因為是針對特定 FQDN, 所以要使用 Application Rule Collection，開放 443 Port

application-github.tf

resource "azurerm_firewall_policy_rule_collection_group" "fprcg-for-transit" {
  application_rule_collection {
    name     = "application-rule-collection-for-transit"
    priority = 300
    action   = "Allow"
    rule {
      name = "application-rule-https-any-to-github"
      protocols {
        type = "Https"
        port = 443
      }
      source_addresses  = ["*"]
      destination_fqdns = ["github.com"]
    }
}

因為 Testinfra 針對網路的部分，比較針對 L3/L4 有著墨，但沒有特別針對 L7 的檢查，故要自己撈 curl -o /dev/null -s -w %{http_code} HTTP Status Code 來做檢查。is_resolvable 主要是用來檢查 DNS 是否正常，若正常則會回傳 IP。

testinfra-github.py

class TestApplicationRules(unittest.TestCase):
    def setUp(self):
        self.host = testinfra.get_host("local://")
    def test_to_github(self):
        # L3/L4 Check
        github_network = self.host.addr("github.com")
        self.assertFalse(github.is_reachable) # Equal to "ping -W 1 -c 1 github.com"
        self.assertTrue(github.port(80).is_reachable) # Equal to "nc -w 1 -z github.com 80"
        self.assertTrue(github.port(443).is_reachable) # Equal to "nc -w 1 -z github.com 443"
        self.assertTrue(github.port(443).is_resolvable) # Equal to "getent ahosts github.com"
        # L7 Check
        github_http = self.host.run("curl --connect-timeout 3 -o /dev/null -s -w %{http_code} https://github.com")
        self.assertNotEqual(github.stdout, "200") # Use HTTP Status Code to check if the website is up

使用者 D 說: 我要能連到 azure.archive.ubuntu.com 進行 apt update && apt upgrade 的操作

Azure 內提供的 azure.archive.ubuntu.com 是給 Ubuntu 系列的 VM 來進行 apt update && apt upgrade 的操作，但僅有 HTTP 沒有 HTTPS，所以這邊也要使用 Application Rule Collection，開放 HTTP/80

application-ubuntu.tf

resource "azurerm_firewall_policy_rule_collection_group" "fprcg-for-transit" {
  application_rule_collection {
    name     = "application-rule-collection-for-transit"
    priority = 301
    action   = "Allow"
    rule {
      name = "application-rule-https-any-to-ubuntu"
      protocols {
        type = "Http"
        port = 80
      }
      source_addresses  = ["*"]
      destination_fqdns = ["azure.archive.ubuntu.com"]
    }
}

這邊針對 HTTPS 不應該存在回應的狀況進行檢查，使用 assertNotEqual 來檢查不應該有 HTTP Status Code 200 的狀況發生

testinfra-ubuntu.py

class TestApplicationRules(unittest.TestCase):
    def setUp(self):
        self.host = testinfra.get_host("local://")
    def test_to_ubuntu(self):
        # L3/L4 Check
        azure_ubuntu_repo = self.host.addr("azure.archive.ubuntu.com")
        self.assertFalse(azure_ubuntu_repo.is_reachable) # Equal to "ping -W 1 -c 1 azure.archive.ubuntu.com"
        self.assertTrue(azure_ubuntu_repo.port(80).is_reachable)  # Equal to "nc -w 1 -z azure.archive.ubuntu.com 80"
        self.assertFalse(azure_ubuntu_repo.port(443).is_reachable) # Equal to "nc -w 1 -z azure.archive.ubuntu.com 443"
        self.assertTrue(azure_ubuntu_repo.is_resolvable) # Equal to "getent ahosts azure.archive.ubuntu.com"
        # L7 Check
        azure_ubuntu_http = self.host.run("curl --connect-timeout 3 -o /dev/null -s -w %{http_code} https://azure.archive.ubuntu.com")
        self.assertNotEqual(azure_ubuntu_http.stdout, "200") # HTTPS is not supported

        azure_ubuntu_http = self.host.run("curl --connect-timeout 3 -o /dev/null -s -w %{http_code} http://azure.archive.ubuntu.com")
        self.assertEqual(azure_ubuntu_http.stdout, "200") # HTTP is supported

Q&A

Q1: 為何不用 Serverspec 來寫測試?

1. Python 在多數的 Linux 環境都有
2. 我個人看不太懂 Ruby 的 Source Code，但 Python 就看得懂了

Q2: pytest 跟 pytest-testinfra 有什麼差別?

• pytest: 幫助你寫出更好的 Python 程式
• pytest-testinfra: 基於 pytest 的基礎，幫助你持續檢查出你沒注意到的 Infrastructure 功能性問題

Q3: 我一定要使用 Azure 才能寫測試嗎?

No. 你此時此刻只要有遇到網路檢查問題，基本上都可以寫

Q4: 我不知道我為什麼被檔了，寫測試可以幫助我嗎?

其實寫 Unit Test 就是幫助你了解你的 Firewall Rule 到底是有正常運作還是沒正常運作，如果你有仔細看 Azure Firewall Log，在上面都會很清楚顯示你這個動作，是對應到哪一條 Policy 而 Allow 或者是 Deny，可以以使用者的角度確保你的認知、使用者的認知以及 Azure Firewall 的設定是正確的

Q5: 我懶得寫，有沒有懶人包

請參考 Code Snippet pichuang/testinfra-code-snippet 複製貼上修改

Q6: 感覺網路測試感覺都是正面表列，都沒有負向表列的嗎?

多使用 assertFalse 和 assertNotEqual 來確保你的測試是正確的，然後你不太可能窮盡所有的可能性，所以一開始以正面表列的 Test Case 先出發，在後續的測試和回饋中，再依照你的需求來增加 Unit Test Case

Q7: 這個有沒有什麼方法論

TDD (Test Driven Development)，只是多半都是用 TDD 幫助程式更好，這邊是用 TDD 幫助你的 Firewall Policy 設計的更好

Q8: 看到 IP 和 URL 建議測試順序為何?

• 你如果是要測試特定 IP 的話，譬如 8.8.8.8，建議測試順序為

  1. ICMP 通不通
  2. 特定 Port 通不通

• 你如果是要測試特定 URL 的話，譬如 https://blog.pichuang.com.tw，建議測試順序為

  1. ICMP 通不通
  2. 特定 Port 通不通
  3. 能不能做 DNS Resolution
  4. 以 Curl 該 URL 獲得 HTTP Status Sode 來判斷正不正常

Q9: 既然都是用 pytest 為主，那幹嘛還要用 unittest 框架?

恩...我一開始就用 unitest 框架寫，然後就...懶得重構了

Q10: 為啥不要全部直接用 GitHub Copilot 自動寫全部測試?

GitHub Copilot 不懂網路資安、法規和人心 :)

References

• Azure Firewall Policy rule sets
• pytest: helps you write better programs
• pichuang/testinfra-code-snippet
• Testinfra: Testinfra test your infrastructure