tools/run_tests/xds_k8s_test_driver/bin/run_channelz.py

   1 # Copyright 2020 gRPC authors.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #     http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14 """Channelz debugging tool for xDS test client/server.
  15
  16 This is intended as a debugging / local development helper and not executed
  17 as a part of interop test suites.
  18
  19 Typical usage examples:
  20
  21     # Show channel and server socket pair
  22     python -m bin.run_channelz --flagfile=config/local-dev.cfg
  23
  24     # Evaluate setup for different security configurations
  25     python -m bin.run_channelz --flagfile=config/local-dev.cfg --security=tls
  26     python -m bin.run_channelz --flagfile=config/local-dev.cfg --security=mtls_error
  27
  28     # More information and usage options
  29     python -m bin.run_channelz --helpful
  30 """
  31 import hashlib
  32 import logging
  33
  34 from absl import app
  35 from absl import flags
  36
  37 from framework import xds_flags
  38 from framework import xds_k8s_flags
  39 from framework.infrastructure import k8s
  40 from framework.rpc import grpc_channelz
  41 from framework.test_app import client_app
  42 from framework.test_app import server_app
  43
  44 logger = logging.getLogger(__name__)
  45 # Flags
  46 _SERVER_RPC_HOST = flags.DEFINE_string('server_rpc_host',
  47                                        default='127.0.0.1',
  48                                        help='Server RPC host')
  49 _CLIENT_RPC_HOST = flags.DEFINE_string('client_rpc_host',
  50                                        default='127.0.0.1',
  51                                        help='Client RPC host')
  52 _SECURITY = flags.DEFINE_enum('security',
  53                               default=None,
  54                               enum_values=[
  55                                   'mtls', 'tls', 'plaintext', 'mtls_error',
  56                                   'server_authz_error'
  57                               ],
  58                               help='Show info for a security setup')
  59 flags.adopt_module_key_flags(xds_flags)
  60 flags.adopt_module_key_flags(xds_k8s_flags)
  61
  62 # Type aliases
  63 _Channel = grpc_channelz.Channel
  64 _Socket = grpc_channelz.Socket
  65 _ChannelState = grpc_channelz.ChannelState
  66 _XdsTestServer = server_app.XdsTestServer
  67 _XdsTestClient = client_app.XdsTestClient
  68
  69
  70 def debug_cert(cert):
  71     if not cert:
  72         return '<missing>'
  73     sha1 = hashlib.sha1(cert)
  74     return f'sha1={sha1.hexdigest()}, len={len(cert)}'
  75
  76
  77 def debug_sock_tls(tls):
  78     return (f'local:  {debug_cert(tls.local_certificate)}\n'
  79             f'remote: {debug_cert(tls.remote_certificate)}')
  80
  81
  82 def get_deployment_pod_ips(k8s_ns, deployment_name):
  83     deployment = k8s_ns.get_deployment(deployment_name)
  84     pods = k8s_ns.list_deployment_pods(deployment)
  85     return [pod.status.pod_ip for pod in pods]
  86
  87
  88 def debug_security_setup_negative(test_client):
  89     """Debug negative cases: mTLS Error, Server AuthZ error
  90
  91     1) mTLS Error: Server expects client mTLS cert,
  92        but client configured only for TLS.
  93     2) AuthZ error: Client does not authorize server because of mismatched
  94        SAN name.
  95     """
  96     # Client side.
  97     client_correct_setup = True
  98     channel: _Channel = test_client.wait_for_server_channel_state(
  99         state=_ChannelState.TRANSIENT_FAILURE)
 100     try:
 101         subchannel, *subchannels = list(
 102             test_client.channelz.list_channel_subchannels(channel))
 103     except ValueError:
 104         print("Client setup fail: subchannel not found. "
 105               "Common causes: test client didn't connect to TD; "
 106               "test client exhausted retries, and closed all subchannels.")
 107         return
 108
 109     # Client must have exactly one subchannel.
 110     logger.debug('Found subchannel, %s', subchannel)
 111     if subchannels:
 112         client_correct_setup = False
 113         print(f'Unexpected subchannels {subchannels}')
 114     subchannel_state: _ChannelState = subchannel.data.state.state
 115     if subchannel_state is not _ChannelState.TRANSIENT_FAILURE:
 116         client_correct_setup = False
 117         print('Subchannel expected to be in '
 118               'TRANSIENT_FAILURE, same as its channel')
 119
 120     # Client subchannel must have no sockets.
 121     sockets = list(test_client.channelz.list_subchannels_sockets(subchannel))
 122     if sockets:
 123         client_correct_setup = False
 124         print(f'Unexpected subchannel sockets {sockets}')
 125
 126     # Results.
 127     if client_correct_setup:
 128         print('Client setup pass: the channel '
 129               'to the server has exactly one subchannel '
 130               'in TRANSIENT_FAILURE, and no sockets')
 131
 132
 133 def debug_security_setup_positive(test_client, test_server):
 134     """Debug positive cases: mTLS, TLS, Plaintext."""
 135     test_client.wait_for_active_server_channel()
 136     client_sock: _Socket = test_client.get_active_server_channel_socket()
 137     server_sock: _Socket = test_server.get_server_socket_matching_client(
 138         client_sock)
 139
 140     server_tls = server_sock.security.tls
 141     client_tls = client_sock.security.tls
 142
 143     print(f'\nServer certs:\n{debug_sock_tls(server_tls)}')
 144     print(f'\nClient certs:\n{debug_sock_tls(client_tls)}')
 145     print()
 146
 147     if server_tls.local_certificate:
 148         eq = server_tls.local_certificate == client_tls.remote_certificate
 149         print(f'(TLS)  Server local matches client remote: {eq}')
 150     else:
 151         print('(TLS)  Not detected')
 152
 153     if server_tls.remote_certificate:
 154         eq = server_tls.remote_certificate == client_tls.local_certificate
 155         print(f'(mTLS) Server remote matches client local: {eq}')
 156     else:
 157         print('(mTLS) Not detected')
 158
 159
 160 def debug_basic_setup(test_client, test_server):
 161     """Show channel and server socket pair"""
 162     test_client.wait_for_active_server_channel()
 163     client_sock: _Socket = test_client.get_active_server_channel_socket()
 164     server_sock: _Socket = test_server.get_server_socket_matching_client(
 165         client_sock)
 166
 167     print(f'Client socket:\n{client_sock}\n')
 168     print(f'Matching server:\n{server_sock}\n')
 169
 170
 171 def main(argv):
 172     if len(argv) > 1:
 173         raise app.UsageError('Too many command-line arguments.')
 174
 175     k8s_api_manager = k8s.KubernetesApiManager(xds_k8s_flags.KUBE_CONTEXT.value)
 176
 177     # Server
 178     server_name = xds_flags.SERVER_NAME.value
 179     server_namespace = xds_flags.NAMESPACE.value
 180     server_k8s_ns = k8s.KubernetesNamespace(k8s_api_manager, server_namespace)
 181     server_pod_ip = get_deployment_pod_ips(server_k8s_ns, server_name)[0]
 182     test_server: _XdsTestServer = _XdsTestServer(
 183         ip=server_pod_ip,
 184         rpc_port=xds_flags.SERVER_PORT.value,
 185         xds_host=xds_flags.SERVER_XDS_HOST.value,
 186         xds_port=xds_flags.SERVER_XDS_PORT.value,
 187         rpc_host=_SERVER_RPC_HOST.value)
 188
 189     # Client
 190     client_name = xds_flags.CLIENT_NAME.value
 191     client_namespace = xds_flags.NAMESPACE.value
 192     client_k8s_ns = k8s.KubernetesNamespace(k8s_api_manager, client_namespace)
 193     client_pod_ip = get_deployment_pod_ips(client_k8s_ns, client_name)[0]
 194     test_client: _XdsTestClient = _XdsTestClient(
 195         ip=client_pod_ip,
 196         server_target=test_server.xds_uri,
 197         rpc_port=xds_flags.CLIENT_PORT.value,
 198         rpc_host=_CLIENT_RPC_HOST.value)
 199
 200     if _SECURITY.value in ('mtls', 'tls', 'plaintext'):
 201         debug_security_setup_positive(test_client, test_server)
 202     elif _SECURITY.value == ('mtls_error', 'server_authz_error'):
 203         debug_security_setup_negative(test_client)
 204     else:
 205         debug_basic_setup(test_client, test_server)
 206
 207     test_client.close()
 208     test_server.close()
 209
 210
 211 if __name__ == '__main__':
 212     app.run(main)