...

Text file src/github.com/datawire/ambassador/v2/python/tests/kat/t_tcpmapping.py

Documentation: github.com/datawire/ambassador/v2/python/tests/kat

     1from typing import Dict, Generator, Literal, Tuple, Union
     2
     3from abstract_tests import HTTP, AmbassadorTest, Node, ServiceType
     4from kat.harness import Query, abstract_test
     5from tests.integration.manifests import namespace_manifest
     6from tests.selfsigned import TLSCerts
     7
     8# An AmbassadorTest subclass will actually create a running Ambassador.
     9# "self" in this class will refer to the Ambassador.
    10
    11
    12class TCPMappingTest(AmbassadorTest):
    13    # single_namespace = True
    14    namespace = "tcp-namespace"
    15    extra_ports = [6789, 7654, 8765, 9876]
    16
    17    # This test is written assuming explicit control of which Hosts are present,
    18    # so don't let Edge Stack mess with that.
    19    edge_stack_cleartext_host = False
    20
    21    # If you set debug = True here, the results of every Query will be printed
    22    # when the test is run.
    23    # debug = True
    24
    25    target1: ServiceType
    26    target2: ServiceType
    27    target3: ServiceType
    28
    29    # init (not __init__) is the method that initializes a KAT Node (including
    30    # Test, AmbassadorTest, etc.).
    31
    32    def init(self):
    33        self.add_default_http_listener = False
    34        self.add_default_https_listener = False
    35
    36        self.target1 = HTTP(name="target1")
    37        # print("TCP target1 %s" % self.target1.namespace)
    38
    39        self.target2 = HTTP(name="target2", namespace="other-namespace")
    40        # print("TCP target2 %s" % self.target2.namespace)
    41
    42        self.target3 = HTTP(name="target3")
    43        # print("TCP target3 %s" % self.target3.namespace)
    44
    45    # manifests returns a string of Kubernetes YAML that will be applied to the
    46    # Kubernetes cluster before running any tests.
    47
    48    def manifests(self) -> str:
    49        return (
    50            namespace_manifest("tcp-namespace")
    51            + namespace_manifest("other-namespace")
    52            + f"""
    53---
    54apiVersion: v1
    55kind: Secret
    56metadata:
    57  name: supersecret
    58type: kubernetes.io/tls
    59data:
    60  tls.crt: {TLSCerts["tls-context-host-2"].k8s_crt}
    61  tls.key: {TLSCerts["tls-context-host-2"].k8s_key}
    62---
    63apiVersion: getambassador.io/v3alpha1
    64kind: Listener
    65metadata:
    66  name: {self.path.k8s}-listener
    67  labels:
    68    kat-ambassador-id: {self.ambassador_id}
    69spec:
    70  ambassador_id: [ "{self.ambassador_id}" ]
    71  port: 8443
    72  protocol: HTTPS
    73  securityModel: XFP
    74  hostBinding:
    75    namespace:
    76      from: ALL
    77---
    78# In most real-world cases, we'd just use a single wildcard Host instead
    79# of using three. For this test, though, we need three because we aren't
    80# using real domain names, and you can't do wildcards like tls-context-*
    81# (because the '*' has to be a domain part on its own).
    82apiVersion: getambassador.io/v3alpha1
    83kind: Host
    84metadata:
    85  name: {self.path.k8s}-host
    86  labels:
    87    kat-ambassador-id: {self.ambassador_id}
    88spec:
    89  ambassador_id: [ "{self.ambassador_id}" ]
    90  hostname: tls-context-host-1
    91  tlsContext:
    92    name: {self.name}-tlscontext
    93  tlsSecret:
    94    name: supersecret
    95  requestPolicy:
    96    insecure:
    97      action: Reject
    98---
    99apiVersion: getambassador.io/v3alpha1
   100kind: Host
   101metadata:
   102  name: {self.path.k8s}-host-2
   103  labels:
   104    kat-ambassador-id: {self.ambassador_id}
   105spec:
   106  ambassador_id: [ "{self.ambassador_id}" ]
   107  hostname: tls-context-host-2
   108  tlsContext:
   109    name: {self.name}-tlscontext
   110  tlsSecret:
   111    name: supersecret
   112  requestPolicy:
   113    insecure:
   114      action: Reject
   115---
   116apiVersion: getambassador.io/v3alpha1
   117kind: Host
   118metadata:
   119  name: {self.path.k8s}-host-3
   120  labels:
   121    kat-ambassador-id: {self.ambassador_id}
   122spec:
   123  ambassador_id: [ "{self.ambassador_id}" ]
   124  hostname: tls-context-host-3
   125  tlsContext:
   126    name: {self.name}-tlscontext
   127  tlsSecret:
   128    name: supersecret
   129  requestPolicy:
   130    insecure:
   131      action: Reject
   132"""
   133            + super().manifests()
   134        )
   135
   136    # config() must _yield_ tuples of Node, Ambassador-YAML where the
   137    # Ambassador-YAML will be annotated onto the Node.
   138
   139    def config(self) -> Generator[Union[str, Tuple[Node, str]], None, None]:
   140        yield self, self.format(
   141            """
   142---
   143apiVersion: getambassador.io/v3alpha1
   144kind: TLSContext
   145name: {self.name}-tlscontext
   146hosts:
   147- tls-context-host-1
   148- tls-context-host-2
   149- tls-context-host-3
   150secret: supersecret
   151"""
   152        )
   153
   154        yield self.target1, self.format(
   155            """
   156---
   157apiVersion: getambassador.io/v3alpha1
   158kind: TCPMapping
   159name:  {self.name}
   160port: 9876
   161service: {self.target1.path.fqdn}:443
   162---
   163apiVersion: getambassador.io/v3alpha1
   164kind: TCPMapping
   165name:  {self.name}-local-only
   166address: 127.0.0.1
   167port: 8765
   168service: {self.target1.path.fqdn}:443
   169---
   170apiVersion: getambassador.io/v3alpha1
   171kind: TCPMapping
   172name:  {self.name}-clear-to-tls
   173port: 7654
   174service: https://{self.target2.path.fqdn}:443
   175---
   176apiVersion: getambassador.io/v3alpha1
   177kind: TCPMapping
   178name:  {self.name}-1
   179port: 6789
   180host: tls-context-host-1
   181service: {self.target1.path.fqdn}:80
   182"""
   183        )
   184
   185        # Host-differentiated.
   186        yield self.target2, self.format(
   187            """
   188---
   189apiVersion: getambassador.io/v3alpha1
   190kind: TCPMapping
   191name:  {self.name}-2
   192port: 6789
   193host: tls-context-host-2
   194service: {self.target2.path.fqdn}
   195tls: {self.name}-tlscontext
   196"""
   197        )
   198
   199        # Host-differentiated.
   200        yield self.target3, self.format(
   201            """
   202---
   203apiVersion: getambassador.io/v3alpha1
   204kind: TCPMapping
   205name:  {self.name}-3
   206port: 6789
   207host: tls-context-host-3
   208service: https://{self.target3.path.fqdn}
   209"""
   210        )
   211
   212    def requirements(self):
   213        # We're replacing super()'s requirements deliberately here. Without a Host header they can't work.
   214        yield (
   215            "url",
   216            Query(
   217                self.url("ambassador/v0/check_ready"),
   218                headers={"Host": "tls-context-host-1"},
   219                insecure=True,
   220                sni=True,
   221            ),
   222        )
   223        yield (
   224            "url",
   225            Query(
   226                self.url("ambassador/v0/check_alive"),
   227                headers={"Host": "tls-context-host-1"},
   228                insecure=True,
   229                sni=True,
   230            ),
   231        )
   232        yield (
   233            "url",
   234            Query(
   235                self.url("ambassador/v0/check_ready"),
   236                headers={"Host": "tls-context-host-2"},
   237                insecure=True,
   238                sni=True,
   239            ),
   240        )
   241        yield (
   242            "url",
   243            Query(
   244                self.url("ambassador/v0/check_alive"),
   245                headers={"Host": "tls-context-host-2"},
   246                insecure=True,
   247                sni=True,
   248            ),
   249        )
   250
   251    # scheme defaults to HTTP; if you need to use HTTPS, have it return
   252    # "https"...
   253    def scheme(self):
   254        return "https"
   255
   256    # Any Query object yielded from queries() will be run as a test. Also,
   257    # you can add a keyword argument debug=True to any Query() call and the
   258    # complete response object will be dumped.
   259
   260    def queries(self):
   261        # 0: should hit target1, and use TLS
   262        yield Query(self.url(self.name + "/wtfo/", port=9876), insecure=True)
   263
   264        # 1: should hit target2, and use TLS
   265        yield Query(self.url(self.name + "/wtfo/", port=7654, scheme="http"), insecure=True)
   266
   267        # 2: should hit target1 via SNI, and use cleartext
   268        yield Query(
   269            self.url(self.name + "/wtfo/", port=6789),
   270            headers={"Host": "tls-context-host-1"},
   271            insecure=True,
   272            sni=True,
   273        )
   274
   275        # 3: should hit target2 via SNI, and use TLS
   276        yield Query(
   277            self.url(self.name + "/wtfo/", port=6789),
   278            headers={"Host": "tls-context-host-2"},
   279            insecure=True,
   280            sni=True,
   281        )
   282
   283        # 4: should hit target3 via SNI, and use TLS
   284        yield Query(
   285            self.url(self.name + "/wtfo/", port=6789),
   286            headers={"Host": "tls-context-host-3"},
   287            insecure=True,
   288            sni=True,
   289        )
   290
   291        # 5: should error since port 8765 is bound only to localhost
   292        yield Query(
   293            self.url(self.name + "/wtfo/", port=8765),
   294            error=["connection reset by peer", "EOF", "connection refused"],
   295            insecure=True,
   296        )
   297
   298    # Once in check(), self.results is an ordered list of results from your
   299    # Queries. (You can also look at self.parent.results if you really want
   300    # to.)
   301
   302    def check(self):
   303        for idx, target, expected_tls in [
   304            (0, self.target1, True),
   305            (1, self.target2, True),
   306            (2, self.target1, False),
   307            (3, self.target2, True),
   308            (4, self.target3, True),
   309            # ( 5, self.target1 ),
   310        ]:
   311            expected_host = target.path.k8s
   312            r = self.results[idx]
   313
   314            assert r.backend
   315            actual_host = r.backend.name
   316            assert r.backend.request
   317            actual_tls = r.backend.request.tls.enabled
   318
   319            assert actual_host == expected_host
   320            assert actual_tls == expected_tls
   321
   322
   323class TCPMappingBasicTest(AmbassadorTest):
   324    extra_ports = [6789]
   325    target: ServiceType
   326
   327    def init(self) -> None:
   328        self.target = HTTP()
   329
   330    def manifests(self) -> str:
   331        return (
   332            format(
   333                """
   334---
   335apiVersion: getambassador.io/v2
   336kind: TCPMapping
   337metadata:
   338  name: {self.path.k8s}
   339spec:
   340  ambassador_id: [ {self.ambassador_id} ]
   341  port: 6789
   342  service: {self.target.path.fqdn}:80
   343"""
   344            )
   345            + super().manifests()
   346        )
   347
   348    def queries(self):
   349        yield Query(self.url("", port=6789))
   350
   351    def check(self):
   352        assert self.results[0].json["backend"] == self.target.path.k8s
   353        assert self.results[0].json["request"]["tls"]["enabled"] == False
   354
   355
   356class TCPMappingCrossNamespaceTest(AmbassadorTest):
   357    extra_ports = [6789]
   358    target: ServiceType
   359
   360    def init(self) -> None:
   361        self.target = HTTP(namespace="other-namespace")
   362
   363    def manifests(self) -> str:
   364        return (
   365            namespace_manifest("other-namespace")
   366            + format(
   367                """
   368---
   369apiVersion: getambassador.io/v2
   370kind: TCPMapping
   371metadata:
   372  name: {self.path.k8s}
   373spec:
   374  ambassador_id: [ {self.ambassador_id} ]
   375  port: 6789
   376  service: {self.target.path.fqdn}:80
   377"""
   378            )
   379            + super().manifests()
   380        )
   381
   382    def queries(self):
   383        yield Query(self.url("", port=6789))
   384
   385    def check(self):
   386        assert self.results[0].json["backend"] == self.target.path.k8s
   387        assert self.results[0].json["request"]["tls"]["enabled"] == False
   388
   389
   390class TCPMappingTLSOriginationBoolTest(AmbassadorTest):
   391    extra_ports = [6789]
   392    target: ServiceType
   393
   394    def init(self) -> None:
   395        self.target = HTTP()
   396
   397    def manifests(self) -> str:
   398        return (
   399            format(
   400                """
   401---
   402apiVersion: getambassador.io/v2
   403kind: TCPMapping
   404metadata:
   405  name: {self.path.k8s}
   406spec:
   407  ambassador_id: [ {self.ambassador_id} ]
   408  port: 6789
   409  service: {self.target.path.fqdn}:443
   410  tls: true
   411"""
   412            )
   413            + super().manifests()
   414        )
   415
   416    def queries(self):
   417        yield Query(self.url("", port=6789))
   418
   419    def check(self):
   420        assert self.results[0].json["backend"] == self.target.path.k8s
   421        assert self.results[0].json["request"]["tls"]["enabled"] == True
   422
   423
   424class TCPMappingTLSOriginationV2SchemeTest(AmbassadorTest):
   425    """apiVersion v2 TCPMappings don't support a scheme:// on the 'service' field; if you provide
   426    one, then it is ignored.  Since apiVersion v3alpha1 adds support for scheme://, add a test to
   427    make sure we don't break anyone who is inadvertently depending on it being ignored in v2."""
   428
   429    extra_ports = [6789, 6790]
   430    target: ServiceType
   431
   432    def init(self) -> None:
   433        self.xfail = "bug (2.3): v2 TCPMappings don't ignore the scheme"
   434        self.target = HTTP()
   435
   436    def manifests(self) -> str:
   437        return (
   438            format(
   439                """
   440---
   441apiVersion: getambassador.io/v2
   442kind: TCPMapping
   443metadata:
   444  name: {self.path.k8s}-1
   445spec:
   446  ambassador_id: [ {self.ambassador_id} ]
   447  port: 6789
   448  service: https://{self.target.path.fqdn}:443
   449---
   450apiVersion: getambassador.io/v2
   451kind: TCPMapping
   452metadata:
   453  name: {self.path.k8s}-2
   454spec:
   455  ambassador_id: [ {self.ambassador_id} ]
   456  port: 6790
   457  service: https://{self.target.path.fqdn}:80
   458"""
   459            )
   460            + super().manifests()
   461        )
   462
   463    def queries(self):
   464        yield Query(
   465            self.url("", port=6789), expected=400
   466        )  # kat-server returns HTTP 400 "Client sent an HTTP request to an HTTPS server."
   467        yield Query(self.url("", port=6789, scheme="https"), insecure=True)
   468        yield Query(self.url("", port=6790))
   469
   470    def check(self):
   471        assert self.results[1].json["backend"] == self.target.path.k8s
   472        assert self.results[1].json["request"]["tls"]["enabled"] == True
   473        assert self.results[2].json["backend"] == self.target.path.k8s
   474        assert self.results[2].json["request"]["tls"]["enabled"] == False
   475
   476
   477class TCPMappingTLSOriginationV3SchemeTest(AmbassadorTest):
   478    extra_ports = [6789]
   479    target: ServiceType
   480
   481    def init(self) -> None:
   482        self.target = HTTP()
   483
   484    def manifests(self) -> str:
   485        return (
   486            format(
   487                """
   488---
   489apiVersion: getambassador.io/v3alpha1
   490kind: TCPMapping
   491metadata:
   492  name: {self.path.k8s}-1
   493spec:
   494  ambassador_id: [ {self.ambassador_id} ]
   495  port: 6789
   496  service: https://{self.target.path.fqdn}:443
   497"""
   498            )
   499            + super().manifests()
   500        )
   501
   502    def queries(self):
   503        yield Query(self.url("", port=6789))
   504
   505    def check(self):
   506        assert self.results[0].json["backend"] == self.target.path.k8s
   507        assert self.results[0].json["request"]["tls"]["enabled"] == True
   508
   509
   510class TCPMappingTLSOriginationContextTest(AmbassadorTest):
   511    extra_ports = [6789]
   512    target: ServiceType
   513
   514    def init(self) -> None:
   515        self.target = HTTP()
   516
   517    def manifests(self) -> str:
   518        # Hafta provide a client cert, see https://github.com/emissary-ingress/emissary/issues/4476
   519        return (
   520            f"""
   521---
   522apiVersion: v1
   523kind: Secret
   524metadata:
   525  name: {self.path.k8s}-clientcert
   526type: kubernetes.io/tls
   527data:
   528  tls.crt: {TLSCerts["presto.example.com"].k8s_crt}
   529  tls.key: {TLSCerts["presto.example.com"].k8s_key}
   530---
   531apiVersion: getambassador.io/v2
   532kind: TLSContext
   533metadata:
   534  name: {self.path.k8s}-tlsclient
   535spec:
   536  ambassador_id: [ {self.ambassador_id} ]
   537  secret: {self.path.k8s}-clientcert
   538  sni: my-funny-name
   539---
   540apiVersion: getambassador.io/v2
   541kind: TCPMapping
   542metadata:
   543  name: {self.path.k8s}
   544spec:
   545  ambassador_id: [ {self.ambassador_id} ]
   546  port: 6789
   547  service: {self.target.path.fqdn}:443
   548  tls: {self.path.k8s}-tlsclient
   549"""
   550            + super().manifests()
   551        )
   552
   553    def queries(self):
   554        yield Query(self.url("", port=6789))
   555
   556    def check(self):
   557        assert self.results[0].json["backend"] == self.target.path.k8s
   558        assert self.results[0].json["request"]["tls"]["enabled"] == True
   559        assert self.results[0].json["request"]["tls"]["server-name"] == "my-funny-name"
   560
   561
   562class TCPMappingTLSOriginationContextWithDotTest(AmbassadorTest):
   563    extra_ports = [6789]
   564    target: ServiceType
   565
   566    def init(self) -> None:
   567        self.target = HTTP()
   568
   569    def manifests(self) -> str:
   570        # Hafta provide a client cert, see https://github.com/emissary-ingress/emissary/issues/4476
   571        return (
   572            f"""
   573---
   574apiVersion: v1
   575kind: Secret
   576metadata:
   577  name: {self.path.k8s}-clientcert
   578type: kubernetes.io/tls
   579data:
   580  tls.crt: {TLSCerts["presto.example.com"].k8s_crt}
   581  tls.key: {TLSCerts["presto.example.com"].k8s_key}
   582---
   583apiVersion: getambassador.io/v2
   584kind: TLSContext
   585metadata:
   586  name: {self.path.k8s}.tlsclient
   587spec:
   588  ambassador_id: [ {self.ambassador_id} ]
   589  secret: {self.path.k8s}-clientcert
   590  sni: my-hilarious-name
   591---
   592apiVersion: getambassador.io/v2
   593kind: TCPMapping
   594metadata:
   595  name: {self.path.k8s}
   596spec:
   597  ambassador_id: [ {self.ambassador_id} ]
   598  port: 6789
   599  service: {self.target.path.fqdn}:443
   600  tls: {self.path.k8s}.tlsclient
   601"""
   602            + super().manifests()
   603        )
   604
   605    def queries(self):
   606        yield Query(self.url("", port=6789))
   607
   608    def check(self):
   609        assert self.results[0].json["backend"] == self.target.path.k8s
   610        assert self.results[0].json["request"]["tls"]["enabled"] == True
   611        assert self.results[0].json["request"]["tls"]["server-name"] == "my-hilarious-name"
   612
   613
   614class TCPMappingTLSOriginationContextCrossNamespaceTest(AmbassadorTest):
   615    """This test is a little funny.  You can actually select a TLSContext from any namespace without
   616    specifying the namespace.  That's bad design, but at the same time we don't want to break anyone
   617    by changing it."""
   618
   619    extra_ports = [6789]
   620    target: ServiceType
   621
   622    def init(self) -> None:
   623        self.target = HTTP()
   624
   625    def manifests(self) -> str:
   626        # Hafta provide a client cert, see https://github.com/emissary-ingress/emissary/issues/4476
   627        return (
   628            namespace_manifest("other-namespace")
   629            + f"""
   630---
   631apiVersion: v1
   632kind: Secret
   633metadata:
   634  name: {self.path.k8s}-clientcert
   635  namespace: other-namespace
   636type: kubernetes.io/tls
   637data:
   638  tls.crt: {TLSCerts["presto.example.com"].k8s_crt}
   639  tls.key: {TLSCerts["presto.example.com"].k8s_key}
   640---
   641apiVersion: getambassador.io/v2
   642kind: TLSContext
   643metadata:
   644  name: {self.path.k8s}-tlsclient
   645  namespace: other-namespace
   646spec:
   647  ambassador_id: [ {self.ambassador_id} ]
   648  secret: {self.path.k8s}-clientcert
   649  sni: my-hysterical-name
   650---
   651apiVersion: getambassador.io/v2
   652kind: TCPMapping
   653metadata:
   654  name: {self.path.k8s}
   655spec:
   656  ambassador_id: [ {self.ambassador_id} ]
   657  port: 6789
   658  service: {self.target.path.fqdn}:443
   659  tls: {self.path.k8s}-tlsclient
   660"""
   661            + super().manifests()
   662        )
   663
   664    def queries(self):
   665        yield Query(self.url("", port=6789))
   666
   667    def check(self):
   668        assert self.results[0].json["backend"] == self.target.path.k8s
   669        assert self.results[0].json["request"]["tls"]["enabled"] == True
   670        assert self.results[0].json["request"]["tls"]["server-name"] == "my-hysterical-name"
   671
   672
   673@abstract_test
   674class TCPMappingTLSTerminationTest(AmbassadorTest):
   675    tls_src: Literal["tlscontext", "host"]
   676
   677    @classmethod
   678    def variants(cls) -> Generator[Node, None, None]:
   679        for tls_src in ["tlscontext", "host"]:
   680            yield cls(tls_src, name="{self.tls_src}")
   681
   682    def init(self, tls_src: Literal["tlscontext", "host"]) -> None:
   683        self.tls_src = tls_src
   684
   685    def manifests(self) -> str:
   686        return (
   687            f"""
   688---
   689apiVersion: getambassador.io/v2
   690kind: Host
   691metadata:
   692  name: {self.path.k8s}
   693spec:
   694  ambassador_id: [ {self.ambassador_id} ]
   695  hostname: {self.path.fqdn}
   696  acmeProvider:
   697    authority: none
   698  requestPolicy:
   699    insecure:
   700      action: Route
   701      additionalPort: 8080
   702"""
   703            + super().manifests()
   704        )
   705
   706
   707class TCPMappingTLSTerminationBasicTest(TCPMappingTLSTerminationTest):
   708    extra_ports = [6789]
   709    target: ServiceType
   710
   711    def init(self, tls_src: Literal["tlscontext", "host"]) -> None:
   712        super().init(tls_src)
   713        self.target = HTTP()
   714
   715    def manifests(self) -> str:
   716        return (
   717            f"""
   718---
   719apiVersion: v1
   720kind: Secret
   721metadata:
   722  name: {self.path.k8s}-servercert
   723type: kubernetes.io/tls
   724data:
   725  tls.crt: {TLSCerts["tls-context-host-2"].k8s_crt}
   726  tls.key: {TLSCerts["tls-context-host-2"].k8s_key}
   727"""
   728            + (
   729                f"""
   730---
   731apiVersion: getambassador.io/v2
   732kind: TLSContext
   733metadata:
   734  name: {self.path.k8s}-tlsserver
   735spec:
   736  ambassador_id: [ {self.ambassador_id} ]
   737  secret: {self.path.k8s}-servercert
   738  hosts: [ "tls-context-host-2" ]
   739"""
   740                if self.tls_src == "tlscontext"
   741                else f"""
   742---
   743apiVersion: getambassador.io/v2
   744kind: Host
   745metadata:
   746  name: {self.path.k8s}-tlsserver
   747spec:
   748  ambassador_id: [ {self.ambassador_id} ]
   749  hostname: "tls-context-host-2"
   750  tlsSecret:
   751    name: {self.path.k8s}-servercert
   752"""
   753            )
   754            + f"""
   755---
   756apiVersion: getambassador.io/v2
   757kind: TCPMapping
   758metadata:
   759  name: {self.path.k8s}
   760spec:
   761  ambassador_id: [ {self.ambassador_id} ]
   762  port: 6789
   763  host: tls-context-host-2
   764  service: {self.target.path.fqdn}:80
   765"""
   766            + super().manifests()
   767        )
   768
   769    def queries(self):
   770        yield Query(
   771            self.url("", scheme="https", port=6789),
   772            sni=True,
   773            headers={"Host": "tls-context-host-2"},
   774            ca_cert=TLSCerts["tls-context-host-2"].pubcert,
   775        )
   776
   777    def check(self):
   778        assert self.results[0].json["backend"] == self.target.path.k8s
   779        assert self.results[0].json["request"]["tls"]["enabled"] == False
   780
   781
   782class TCPMappingTLSTerminationCrossNamespaceTest(TCPMappingTLSTerminationTest):
   783    extra_ports = [6789]
   784    target: ServiceType
   785
   786    def init(self, tls_src: Literal["tlscontext", "host"]) -> None:
   787        super().init(tls_src)
   788        self.target = HTTP()
   789
   790    def manifests(self) -> str:
   791        return (
   792            namespace_manifest("other-namespace")
   793            + f"""
   794---
   795apiVersion: v1
   796kind: Secret
   797metadata:
   798  name: {self.path.k8s}-servercert
   799  namespace: other-namespace
   800type: kubernetes.io/tls
   801data:
   802  tls.crt: {TLSCerts["tls-context-host-2"].k8s_crt}
   803  tls.key: {TLSCerts["tls-context-host-2"].k8s_key}
   804"""
   805            + (
   806                f"""
   807---
   808apiVersion: getambassador.io/v2
   809kind: TLSContext
   810metadata:
   811  name: {self.path.k8s}-tlsserver
   812  namespace: other-namespace
   813spec:
   814  ambassador_id: [ {self.ambassador_id} ]
   815  secret: {self.path.k8s}-servercert
   816  hosts: [ "tls-context-host-2" ]
   817"""
   818                if self.tls_src == "tlscontext"
   819                else f"""
   820---
   821apiVersion: getambassador.io/v2
   822kind: Host
   823metadata:
   824  name: {self.path.k8s}-tlsserver
   825  namespace: other-namespace
   826spec:
   827  ambassador_id: [ {self.ambassador_id} ]
   828  hostname: "tls-context-host-2"
   829  tlsSecret:
   830    name: {self.path.k8s}-servercert
   831"""
   832            )
   833            + f"""
   834---
   835apiVersion: getambassador.io/v2
   836kind: TCPMapping
   837metadata:
   838  name: {self.path.k8s}
   839spec:
   840  ambassador_id: [ {self.ambassador_id} ]
   841  port: 6789
   842  host: tls-context-host-2
   843  service: {self.target.path.fqdn}:80
   844"""
   845            + super().manifests()
   846        )
   847
   848    def queries(self):
   849        yield Query(
   850            self.url("", scheme="https", port=6789),
   851            sni=True,
   852            headers={"Host": "tls-context-host-2"},
   853            ca_cert=TLSCerts["tls-context-host-2"].pubcert,
   854        )
   855
   856    def check(self):
   857        assert self.results[0].json["backend"] == self.target.path.k8s
   858        assert self.results[0].json["request"]["tls"]["enabled"] == False
   859
   860
   861class TCPMappingSNISharedContextTest(TCPMappingTLSTerminationTest):
   862    extra_ports = [6789]
   863    target_a: ServiceType
   864    target_b: ServiceType
   865
   866    def init(self, tls_src: Literal["tlscontext", "host"]) -> None:
   867        super().init(tls_src)
   868        self.target_a = HTTP(name="target-a")
   869        self.target_b = HTTP(name="target-b")
   870
   871    def manifests(self) -> str:
   872        # Note that TCPMapping.spec.host matches with TLSContext.spec.hosts based on simple string
   873        # matching, not globbing.  See irbasemapping.py:match_tls_context()
   874        return (
   875            f"""
   876---
   877apiVersion: v1
   878kind: Secret
   879metadata:
   880  name: {self.path.k8s}-servercert
   881type: kubernetes.io/tls
   882data:
   883  tls.crt: {TLSCerts["*.domain.com"].k8s_crt}
   884  tls.key: {TLSCerts["*.domain.com"].k8s_key}
   885"""
   886            + (
   887                f"""
   888---
   889apiVersion: getambassador.io/v2
   890kind: TLSContext
   891metadata:
   892  name: {self.path.k8s}-tlsserver
   893spec:
   894  ambassador_id: [ {self.ambassador_id} ]
   895  secret: {self.path.k8s}-servercert
   896  hosts:
   897    - "a.domain.com"
   898    - "b.domain.com"
   899"""
   900                if self.tls_src == "tlscontext"
   901                else f"""
   902---
   903apiVersion: getambassador.io/v2
   904kind: Host
   905metadata:
   906  name: {self.path.k8s}-tlsserver
   907spec:
   908  ambassador_id: [ {self.ambassador_id} ]
   909  hostname: "*.domain.com"
   910  tlsSecret:
   911    name: {self.path.k8s}-servercert
   912  requestPolicy:
   913    insecure:
   914      action: Route
   915      additionalPort: 8080
   916"""
   917            )
   918            + f"""
   919---
   920apiVersion: getambassador.io/v2
   921kind: TCPMapping
   922metadata:
   923  name: {self.path.k8s}-a
   924spec:
   925  ambassador_id: [ {self.ambassador_id} ]
   926  port: 6789
   927  host: a.domain.com
   928  service: {self.target_a.path.fqdn}:80
   929---
   930apiVersion: getambassador.io/v2
   931kind: TCPMapping
   932metadata:
   933  name: {self.path.k8s}-b
   934spec:
   935  ambassador_id: [ {self.ambassador_id} ]
   936  port: 6789
   937  host: b.domain.com
   938  service: {self.target_b.path.fqdn}:80
   939"""
   940            + super().manifests()
   941        )
   942
   943    def queries(self):
   944        yield Query(
   945            self.url("", scheme="https", port=6789),
   946            sni=True,
   947            headers={"Host": "a.domain.com"},
   948            ca_cert=TLSCerts["*.domain.com"].pubcert,
   949        )
   950        yield Query(
   951            self.url("", scheme="https", port=6789),
   952            sni=True,
   953            headers={"Host": "b.domain.com"},
   954            ca_cert=TLSCerts["*.domain.com"].pubcert,
   955        )
   956
   957    def check(self):
   958        assert self.results[0].json["backend"] == self.target_a.path.k8s
   959        assert self.results[0].json["request"]["tls"]["enabled"] == False
   960        assert self.results[1].json["backend"] == self.target_b.path.k8s
   961        assert self.results[1].json["request"]["tls"]["enabled"] == False
   962
   963
   964class TCPMappingSNISeparateContextsTest(TCPMappingTLSTerminationTest):
   965    extra_ports = [6789]
   966    target_a: ServiceType
   967    target_b: ServiceType
   968
   969    def init(self, tls_src: Literal["tlscontext", "host"]) -> None:
   970        super().init(tls_src)
   971        self.target_a = HTTP(name="target-a")
   972        self.target_b = HTTP(name="target-b")
   973
   974    def manifests(self) -> str:
   975        return (
   976            f"""
   977---
   978apiVersion: v1
   979kind: Secret
   980metadata:
   981  name: {self.path.k8s}-servercert-a
   982type: kubernetes.io/tls
   983data:
   984  tls.crt: {TLSCerts["tls-context-host-1"].k8s_crt}
   985  tls.key: {TLSCerts["tls-context-host-1"].k8s_key}
   986---
   987apiVersion: v1
   988kind: Secret
   989metadata:
   990  name: {self.path.k8s}-servercert-b
   991type: kubernetes.io/tls
   992data:
   993  tls.crt: {TLSCerts["tls-context-host-2"].k8s_crt}
   994  tls.key: {TLSCerts["tls-context-host-2"].k8s_key}
   995"""
   996            + (
   997                f"""
   998---
   999apiVersion: getambassador.io/v2
  1000kind: TLSContext
  1001metadata:
  1002  name: {self.path.k8s}-tlsserver-a
  1003spec:
  1004  ambassador_id: [ {self.ambassador_id} ]
  1005  secret: {self.path.k8s}-servercert-a
  1006  hosts: [tls-context-host-1]
  1007---
  1008apiVersion: getambassador.io/v2
  1009kind: TLSContext
  1010metadata:
  1011  name: {self.path.k8s}-tlsserver-b
  1012spec:
  1013  ambassador_id: [ {self.ambassador_id} ]
  1014  secret: {self.path.k8s}-servercert-b
  1015  hosts: [tls-context-host-2]
  1016"""
  1017                if self.tls_src == "tlscontext"
  1018                else f"""
  1019---
  1020apiVersion: getambassador.io/v2
  1021kind: Host
  1022metadata:
  1023  name: {self.path.k8s}-tlsserver-a
  1024spec:
  1025  ambassador_id: [ {self.ambassador_id} ]
  1026  hostname: "tls-context-host-1"
  1027  tlsSecret:
  1028    name: {self.path.k8s}-servercert-a
  1029---
  1030apiVersion: getambassador.io/v2
  1031kind: Host
  1032metadata:
  1033  name: {self.path.k8s}-tlsserver-b
  1034spec:
  1035  ambassador_id: [ {self.ambassador_id} ]
  1036  hostname: "tls-context-host-2"
  1037  tlsSecret:
  1038    name: {self.path.k8s}-servercert-b
  1039"""
  1040            )
  1041            + f"""
  1042---
  1043apiVersion: getambassador.io/v2
  1044kind: TCPMapping
  1045metadata:
  1046  name: {self.path.k8s}-a
  1047spec:
  1048  ambassador_id: [ {self.ambassador_id} ]
  1049  port: 6789
  1050  host: tls-context-host-1
  1051  service: {self.target_a.path.fqdn}:80
  1052---
  1053apiVersion: getambassador.io/v2
  1054kind: TCPMapping
  1055metadata:
  1056  name: {self.path.k8s}-b
  1057spec:
  1058  ambassador_id: [ {self.ambassador_id} ]
  1059  port: 6789
  1060  host: tls-context-host-2
  1061  service: {self.target_b.path.fqdn}:80
  1062"""
  1063            + super().manifests()
  1064        )
  1065
  1066    def queries(self):
  1067        yield Query(
  1068            self.url("", scheme="https", port=6789),
  1069            sni=True,
  1070            headers={"Host": "tls-context-host-1"},
  1071            ca_cert=TLSCerts["tls-context-host-1"].pubcert,
  1072        )
  1073        yield Query(
  1074            self.url("", scheme="https", port=6789),
  1075            sni=True,
  1076            headers={"Host": "tls-context-host-2"},
  1077            ca_cert=TLSCerts["tls-context-host-2"].pubcert,
  1078        )
  1079
  1080    def check(self):
  1081        assert self.results[0].json["backend"] == self.target_a.path.k8s
  1082        assert self.results[0].json["request"]["tls"]["enabled"] == False
  1083        assert self.results[1].json["backend"] == self.target_b.path.k8s
  1084        assert self.results[1].json["request"]["tls"]["enabled"] == False
  1085
  1086
  1087class TCPMappingSNIWithHTTPTest(AmbassadorTest):
  1088    # Note: TCPMappingSNIWithHTTPTest does *not* inherit from TCPMappingTLSTerminationTest because
  1089    # TCPMappingSNIWithHTTPTest wants to take more ownership of the HTTP Host.
  1090
  1091    target: ServiceType
  1092
  1093    tls_src: Literal["tlscontext", "host"]
  1094
  1095    @classmethod
  1096    def variants(cls) -> Generator[Node, None, None]:
  1097        for tls_src in ["tlscontext", "host"]:
  1098            yield cls(tls_src, name="{self.tls_src}")
  1099
  1100    def init(self, tls_src: Literal["tlscontext", "host"]) -> None:
  1101        self.tls_src = tls_src
  1102        self.target = HTTP()
  1103
  1104    def manifests(self) -> str:
  1105        return (
  1106            f"""
  1107# HTTP Host ##########################################################
  1108---
  1109apiVersion: v1
  1110kind: Secret
  1111metadata:
  1112  name: {self.path.k8s}
  1113type: kubernetes.io/tls
  1114data:
  1115  tls.crt: {TLSCerts["tls-context-host-1"].k8s_crt}
  1116  tls.key: {TLSCerts["tls-context-host-1"].k8s_key}
  1117---
  1118apiVersion: getambassador.io/v2
  1119kind: Host
  1120metadata:
  1121  name: {self.path.k8s}
  1122spec:
  1123  ambassador_id: [ {self.ambassador_id} ]
  1124  hostname: {self.path.fqdn}
  1125  acmeProvider:
  1126    authority: none
  1127  tlsSecret:
  1128    name: {self.path.k8s}
  1129# TCPMapping #########################################################
  1130---
  1131apiVersion: v1
  1132kind: Secret
  1133metadata:
  1134  name: {self.path.k8s}-servercert
  1135type: kubernetes.io/tls
  1136data:
  1137  tls.crt: {TLSCerts["tls-context-host-2"].k8s_crt}
  1138  tls.key: {TLSCerts["tls-context-host-2"].k8s_key}
  1139"""
  1140            + (
  1141                f"""
  1142---
  1143apiVersion: getambassador.io/v2
  1144kind: TLSContext
  1145metadata:
  1146  name: {self.path.k8s}-tlsserver
  1147spec:
  1148  ambassador_id: [ {self.ambassador_id} ]
  1149  secret: {self.path.k8s}-servercert
  1150  hosts: [ "tls-context-host-2" ]
  1151"""
  1152                if self.tls_src == "tlscontext"
  1153                else f"""
  1154---
  1155apiVersion: getambassador.io/v2
  1156kind: Host
  1157metadata:
  1158  name: {self.path.k8s}-tlsserver
  1159spec:
  1160  ambassador_id: [ {self.ambassador_id} ]
  1161  hostname: "tls-context-host-2"
  1162  tlsSecret:
  1163    name: {self.path.k8s}-servercert
  1164"""
  1165            )
  1166            + f"""
  1167---
  1168apiVersion: getambassador.io/v2
  1169kind: TCPMapping
  1170metadata:
  1171  name: {self.path.k8s}
  1172spec:
  1173  ambassador_id: [ {self.ambassador_id} ]
  1174  port: 8443
  1175  host: tls-context-host-2
  1176  service: {self.target.path.fqdn}:80
  1177"""
  1178            + super().manifests()
  1179        )
  1180
  1181    def scheme(self):
  1182        return "https"
  1183
  1184    def queries(self):
  1185        yield Query(
  1186            self.url(""),
  1187            sni=True,
  1188            headers={"Host": "tls-context-host-2"},
  1189            ca_cert=TLSCerts["tls-context-host-2"].pubcert,
  1190        )
  1191
  1192    def check(self):
  1193        assert self.results[0].json["backend"] == self.target.path.k8s
  1194        assert self.results[0].json["request"]["tls"]["enabled"] == False
  1195
  1196
  1197class TCPMappingAddressTest(AmbassadorTest):
  1198    extra_ports = [6789, 6790]
  1199    target: ServiceType
  1200
  1201    def init(self) -> None:
  1202        self.target = HTTP()
  1203
  1204    def manifests(self) -> str:
  1205        return (
  1206            format(
  1207                """
  1208---
  1209apiVersion: getambassador.io/v2
  1210kind: TCPMapping
  1211metadata:
  1212  name: {self.path.k8s}-local-only
  1213spec:
  1214  ambassador_id: [ {self.ambassador_id} ]
  1215  port: 6789
  1216  address: 127.0.0.1
  1217  service: {self.target.path.fqdn}:80
  1218---
  1219apiVersion: getambassador.io/v2
  1220kind: TCPMapping
  1221metadata:
  1222  name: {self.path.k8s}-proxy
  1223spec:
  1224  ambassador_id: [ {self.ambassador_id} ]
  1225  port: 6790
  1226  service: localhost:6789
  1227"""
  1228            )
  1229            + super().manifests()
  1230        )
  1231
  1232    def queries(self):
  1233        # Check that it only bound to localhost and doesn't allow external connections.
  1234        yield Query(
  1235            self.url("", port=6789), error=["connection reset by peer", "EOF", "connection refused"]
  1236        )
  1237        # Use a second mapping that proxies to the first to check that it was even created.
  1238        yield Query(self.url("", port=6790))
  1239
  1240    def check(self):
  1241        assert self.results[1].json["backend"] == self.target.path.k8s
  1242        assert self.results[1].json["request"]["tls"]["enabled"] == False
  1243
  1244
  1245class TCPMappingWeightTest(AmbassadorTest):
  1246    extra_ports = [6789]
  1247    target70: ServiceType
  1248    target30: ServiceType
  1249
  1250    def init(self) -> None:
  1251        self.target70 = HTTP(name="tgt70")
  1252        self.target30 = HTTP(name="tgt30")
  1253
  1254    def manifests(self) -> str:
  1255        return (
  1256            format(
  1257                """
  1258---
  1259apiVersion: getambassador.io/v2
  1260kind: TCPMapping
  1261metadata:
  1262  name: {self.path.k8s}-70
  1263spec:
  1264  ambassador_id: [ {self.ambassador_id} ]
  1265  port: 6789
  1266  service: {self.target70.path.fqdn}:80
  1267  weight: 70
  1268---
  1269apiVersion: getambassador.io/v2
  1270kind: TCPMapping
  1271metadata:
  1272  name: {self.path.k8s}-30
  1273spec:
  1274  ambassador_id: [ {self.ambassador_id} ]
  1275  port: 6789
  1276  service: {self.target30.path.fqdn}:80
  1277  weight: 30
  1278"""
  1279            )
  1280            + super().manifests()
  1281        )
  1282
  1283    def queries(self):
  1284        for i in range(1000):
  1285            yield Query(self.url("", port=6789))
  1286
  1287    def check(self):
  1288        counts: Dict[str, int] = {}
  1289        for result in self.results:
  1290            backend = result.json["backend"]
  1291            counts[backend] = counts.get(backend, 0) + 1
  1292        assert counts[self.target70.path.k8s] + counts[self.target30.path.k8s] == 1000
  1293        # Probabalistic, margin might need tuned
  1294        margin = 150
  1295        assert abs(counts[self.target70.path.k8s] - 700) < margin
  1296        assert abs(counts[self.target30.path.k8s] - 300) < margin
  1297
  1298
  1299# TODO: Add tests for all of the config knobs for the upstream connection:
  1300#  - enable_ipv4: false
  1301#  - enable_ipv6: false
  1302#  - circuit_breakers
  1303#  - idle_timeout_ms
  1304#  - resolver
  1305#
  1306# TODO: Add tests for the config knobs for stats:
  1307#  - cluster_tag
  1308#  - stats_name (v3alpha1 only)

View as plain text