code: def _tunnel_task(
      sess: Session, task_id: str, port_map: dict[int, int]
  ) -> Iterator[None]:
      # Args:
      #   port_map: dict of local port => task port.
      #   task_id: tunneled task_id.
  
      master_addr = sess._master
      listeners = [
          ListenerConfig(
              service_id=f"{task_id}:{task_port}", local_port=local_port
          )
          for local_port, task_port in port_map.items()
      ]
      cert = sess._cert
      cert_file, cert_name = None, None
      if cert is not None:
          cert_file = cert.bundle
          cert_name = cert.name
  
      token = authentication.must_cli_auth().get_session_token()
  
      with http_tunnel_listener(
          master_addr, listeners, cert_file, cert_name, token
      ):
          yield
  
imports: from __future__ import annotations
  import contextlib
  import io
  import os
  import socket
  import socketserver
  import ssl
  import sys
  import threading
  import time
  import urllib.request
  from dataclasses import dataclass
  from typing import Iterator, List, Optional, Union
  import lomond
  from determined.common.api import Session, authentication, bindings, request
label_desc: Use image node_id 8 for calling a global function with example usage: _tunnel_task(sess, task_id, port_map) without return types