Create a gRPC channel to the gateway with mTLS transport.
()
| 128 | |
| 129 | |
| 130 | def _grpc_channel() -> grpc.Channel: |
| 131 | """Create a gRPC channel to the gateway with mTLS transport.""" |
| 132 | endpoint, is_tls = _gateway_endpoint() |
| 133 | parsed = urllib.parse.urlparse(endpoint) |
| 134 | host = parsed.hostname or "127.0.0.1" |
| 135 | port = parsed.port or (443 if is_tls else 80) |
| 136 | target = f"{host}:{port}" |
| 137 | |
| 138 | if is_tls: |
| 139 | mtls = _mtls_dir() |
| 140 | ca_cert = (mtls / "ca.crt").read_bytes() |
| 141 | client_cert = (mtls / "tls.crt").read_bytes() |
| 142 | client_key = (mtls / "tls.key").read_bytes() |
| 143 | creds = grpc.ssl_channel_credentials( |
| 144 | root_certificates=ca_cert, |
| 145 | private_key=client_key, |
| 146 | certificate_chain=client_cert, |
| 147 | ) |
| 148 | return grpc.secure_channel(target, creds) |
| 149 | return grpc.insecure_channel(target) |
| 150 | |
| 151 | |
| 152 | def _stub_with_token(token: str) -> tuple[openshell_pb2_grpc.OpenShellStub, list[tuple[str, str]]]: |
no test coverage detected