[action] [PR:168] Adding support for wildcard CN matching during client cert authentication

azurepipeline.yml

@@ -24,7 +24,7 @@ jobs:
     - script: |
         set -ex
         ./build.sh
-        docker run -d --rm -p8090:8090 -p6379:6379 --name rest-api --cap-add NET_ADMIN --privileged -t rest-api-image-test_local:latest
+        docker run -d --rm -p8090:8090 -p6379:6379 -p8081:8081 --name rest-api --cap-add NET_ADMIN --privileged -t rest-api-image-test_local:latest
         docker save rest-api-image-test_local | gzip > rest-api-image-test_local.gz
         docker save rest-api-build-image | gzip > rest-api-build-image.gz
         cp *.gz $(Build.ArtifactStagingDirectory)

go-server-server/go/auth.go

@@ -1,24 +1,32 @@
 package restapi
 
 import (
-    "log"
-    "net/http"
+	"log"
+	"net/http"
+	"strings"
 )
 
 func CommonNameMatch(r *http.Request) bool {
-    //FIXME : in the authentication of client certificate,  after the certificate chain is validated by 
-    // TLS, here we will futher check if the common name of the end-entity certificate is in the trusted 
-	// common name list of the server config. A more strict check may be added here later.
+    // During client cert authentication, after the certificate chain is validated by
+    // TLS, here we will further check if at least one of the common names in the end-entity certificate
+	// matches one of the trusted common names of the server config.
 	for _, peercert := range r.TLS.PeerCertificates {
 		commonName := peercert.Subject.CommonName
+		log.Printf("info: CommonName in the client cert: %s", commonName)
 		for _, name := range trustedCertCommonNames {
-			if commonName == name {
+			if strings.HasPrefix(name, "*") {
+				// wildcard common name matching
+				domain := name[1:]  //strip "*"
+				if strings.HasSuffix(commonName, domain) {
+					log.Printf("info: CommonName %s in the client cert matches trusted wildcard common name %s", commonName, name)
+					return true;
+				}
+			} else if commonName == name {
 				return true;
 			}
 		}
-		log.Printf("info: CommonName in the client cert: %s", commonName)
 	}
 
-    log.Printf("error: Authentication Fail! None of the CommonNames in the client cert are found in trusted common names")
+    log.Printf("error: Authentication Fail! None of the CommonNames in the client cert match any of the trusted common names")
     return false;
 }

supervisor/rest_api_test.conf

@@ -1,5 +1,5 @@
 [program:rest-api]
-command=/usr/sbin/go-server-server.test -test.coverprofile=/coverage.cov -systemtest=true -enablehttps=true -clientcert=/usr/sbin/cert/client/selfsigned.crt -servercert=/usr/sbin/cert/server/selfsigned.crt -serverkey=/usr/sbin/cert/server/selfsigned.key -localapitestdocker=true -loglevel trace
+command=/usr/sbin/go-server-server.test -test.coverprofile=/coverage.cov -systemtest=true -enablehttps=true -clientcert=/usr/sbin/cert/client/selfsigned.crt -servercert=/usr/sbin/cert/server/selfsigned.crt -serverkey=/usr/sbin/cert/server/selfsigned.key -localapitestdocker=true -clientcertcommonname=test.client.restapi.sonic,*.example.sonic,*test.sonic -loglevel trace
 priority=1
 autostart=false
 autorestart=false

test/restapi_client.py

@@ -11,6 +11,7 @@
 VLAN_NAME_PREF    = "Vlan"
 
 TEST_HOST = 'http://localhost:8090/'
+TEST_HOST_HTTPS = "https://localhost:8081/"
 
 class RESTAPI_client:
 
@@ -45,15 +46,21 @@ def patch(self, url, body = []):
         logging.info('Response Body: %s' % r.text)
         return r
 
-    def get(self, url, body = [], params = {}):
+    def get(self, url, body = [], params = {}, client_cert=None):
+        """
+        :param client_cert: tuple of (cert_file, key_file) for client certificate authentication
+        """
         if body == None:
             data = None
         else:
             data = json.dumps(body)
 
         logging.info("Request GET: %s" % url)
         logging.info("JSON Body: %s" % data)
-        r = requests.get(TEST_HOST + url, data=data, params=params, headers={'Content-Type': 'application/json'})
+        if client_cert:
+            r = requests.get(TEST_HOST_HTTPS + url, data=data, params=params, headers={'Content-Type': 'application/json'}, cert=client_cert, verify=False)
+        else:
+            r = requests.get(TEST_HOST + url, data=data, params=params, headers={'Content-Type': 'application/json'})
         logging.info('Response Code: %s' % r.status_code)
         logging.info('Response Body: %s' % r.text)
         return r
@@ -71,6 +78,9 @@ def delete(self, url, body = [], params = {}):
         logging.info('Response Body: %s' % r.text)
         return r
 
+    def get_heartbeat(self, client_cert=None):
+        return self.get("v1/state/heartbeat", client_cert=client_cert)
+
     def get_config_reset_status(self):
         return self.get('v1/config/resetstatus')
 

test/test_restapi.py

@@ -2,6 +2,7 @@
 
 import logging
 import json
+import os
 
 # DB Names
 VXLAN_TUNNEL_TB   = "VXLAN_TUNNEL"
@@ -29,6 +30,128 @@
 def sorted(l):
     return sorted_org(l, key = lambda x : str(sorted_org(x.items())) if isinstance(x, dict) else x)
 
+
+class ClientCert:
+    def __init__(self, common_name, cert_name="restapiclient"):
+        self.common_name = common_name
+        self.cert_name = cert_name
+
+    def __enter__(self):
+        """
+        Generate a client certificate using self.common_name.
+        This new certificate is signed by the self-signed certificate in cert/client/selfsigned.crt.
+        """
+        self.cert = f"{self.cert_name}.crt"
+        self.key = f"{self.cert_name}.key"
+        self.csr = f"{self.cert_name}.csr"
+        os.system(f"openssl genrsa -out {self.key} 2048")
+        os.system(f"openssl req -new -key {self.key} -subj '/CN={self.common_name}' -out {self.csr}")
+        os.system(f"openssl x509 -req -in {self.csr} -CA ../cert/client/selfsigned.crt \
+                    -CAkey ../cert/client/selfsigned.key -CAcreateserial -out {self.cert} -days 825 -sha256")
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        """
+        Clean up generated cert, key, and csr files.
+        """
+        os.remove(self.cert)
+        os.remove(self.csr)
+        os.remove(self.key)
+
+
+class TestClientCertAuth:
+
+    # Exact match tests for "test.client.restapi.sonic"
+
+    def test_exact_match_success(self, setup_restapi_client):
+        _, _, _, restapi_client = setup_restapi_client
+        with ClientCert("test.client.restapi.sonic") as client_cert:
+            r = restapi_client.get_heartbeat(client_cert=(client_cert.cert, client_cert.key))
+            assert r.status_code == 200
+
+    def test_exact_match_failure(self, setup_restapi_client):
+        _, _, _, restapi_client = setup_restapi_client
+        with ClientCert("client.restapi.sonic") as client_cert:
+            r = restapi_client.get_heartbeat(client_cert=(client_cert.cert, client_cert.key))
+            assert r.status_code == 401
+
+    # Wildcard match tests for "*.example.sonic"
+
+    def test_wildcard_match_success_1(self, setup_restapi_client):
+        _, _, _, restapi_client = setup_restapi_client
+        with ClientCert("test.example.sonic") as client_cert:
+            r = restapi_client.get_heartbeat(client_cert=(client_cert.cert, client_cert.key))
+            assert r.status_code == 200
+
+    def test_wildcard_match_success_2(self, setup_restapi_client):
+        _, _, _, restapi_client = setup_restapi_client
+        with ClientCert("another.test.example.sonic") as client_cert:
+            r = restapi_client.get_heartbeat(client_cert=(client_cert.cert, client_cert.key))
+            assert r.status_code == 200
+
+    def test_wildcard_match_failure_1(self, setup_restapi_client):
+        _, _, _, restapi_client = setup_restapi_client
+        with ClientCert("example.sonic") as client_cert:
+            r = restapi_client.get_heartbeat(client_cert=(client_cert.cert, client_cert.key))
+            assert r.status_code == 401
+
+    def test_wildcard_match_failure_2(self, setup_restapi_client):
+        _, _, _, restapi_client = setup_restapi_client
+        with ClientCert("test.example") as client_cert:
+            r = restapi_client.get_heartbeat(client_cert=(client_cert.cert, client_cert.key))
+            assert r.status_code == 401
+
+    def test_wildcard_match_failure_3(self, setup_restapi_client):
+        _, _, _, restapi_client = setup_restapi_client
+        with ClientCert("test.example.com") as client_cert:
+            r = restapi_client.get_heartbeat(client_cert=(client_cert.cert, client_cert.key))
+            assert r.status_code == 401
+
+    def test_wildcard_match_failure_4(self, setup_restapi_client):
+        _, _, _, restapi_client = setup_restapi_client
+        with ClientCert("someexample.sonic") as client_cert:
+            r = restapi_client.get_heartbeat(client_cert=(client_cert.cert, client_cert.key))
+            assert r.status_code == 401
+
+    def test_wildcard_match_failure_5(self, setup_restapi_client):
+        _, _, _, restapi_client = setup_restapi_client
+        with ClientCert("test.example.sonic.com") as client_cert:
+            r = restapi_client.get_heartbeat(client_cert=(client_cert.cert, client_cert.key))
+            assert r.status_code == 401
+
+    # Wildcard match tests for "*test.sonic"
+
+    def test_wildcard_match_success_a(self, setup_restapi_client):
+        _, _, _, restapi_client = setup_restapi_client
+        with ClientCert("mytest.sonic") as client_cert:
+            r = restapi_client.get_heartbeat(client_cert=(client_cert.cert, client_cert.key))
+            assert r.status_code == 200
+
+    def test_wildcard_match_success_b(self, setup_restapi_client):
+        _, _, _, restapi_client = setup_restapi_client
+        with ClientCert("test.sonic") as client_cert:
+            r = restapi_client.get_heartbeat(client_cert=(client_cert.cert, client_cert.key))
+            assert r.status_code == 200
+
+    def test_wildcard_match_success_c(self, setup_restapi_client):
+        _, _, _, restapi_client = setup_restapi_client
+        with ClientCert("example.test.sonic") as client_cert:
+            r = restapi_client.get_heartbeat(client_cert=(client_cert.cert, client_cert.key))
+            assert r.status_code == 200
+
+    def test_wildcard_match_failure_a(self, setup_restapi_client):
+        _, _, _, restapi_client = setup_restapi_client
+        with ClientCert("est.sonic") as client_cert:
+            r = restapi_client.get_heartbeat(client_cert=(client_cert.cert, client_cert.key))
+            assert r.status_code == 401
+
+    def test_wildcard_match_failure_b(self, setup_restapi_client):
+        _, _, _, restapi_client = setup_restapi_client
+        with ClientCert("test.sonico") as client_cert:
+            r = restapi_client.get_heartbeat(client_cert=(client_cert.cert, client_cert.key))
+            assert r.status_code == 401
+
+
 class TestRestApiPositive:
     """Normal behaviour tests"""
     # Helper func