<commit_before># ==================================================================================================
# Copyright 2012 Twitter, Inc.
# --------------------------------------------------------------------------------------------------
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this work except in compliance with the License.
# You may obtain a copy of the License in the LICENSE file, or at:
#
#  http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ==================================================================================================

import atexit
import errno
import socket
import subprocess
import time

from twitter.common import log
from twitter.common.quantity import Amount, Time

try:
  from twitter.common import app
  HAS_APP=True

  app.add_option(
    '--tunnel_host',
    type='string',
    dest='tunnel_host',
    default='nest1.corp.twitter.com',
    help='Host to tunnel commands through (default: %default)')

except ImportError:
  HAS_APP=False


__all__ = (
  'TunnelHelper',
)


def safe_kill(po):
  """
    Given a Popen object, safely kill it without an unexpected exception.
  """
  try:
    po.kill()
  except OSError as e:
    if e.errno != errno.ESRCH:
      raise
  po.wait()


# TODO(wickman) Add mox tests for this.

class TunnelHelper(object):
  """ Class to initiate an SSH or SOCKS tunnel to a remote host through a tunnel host.

  The ssh binary must be on the PATH.
  """
  TUNNELS = {}
  PROXIES = {}
  MIN_RETRY = Amount(5, Time.MILLISECONDS)

  class TunnelError(Exception): pass

  @classmethod
  def log(cls, msg):
    log.debug('%s: %s' % (cls.__name__, msg))

  @classmethod
  def get_random_port(cls):
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    s.bind(('localhost', 0))
    _, port = s.getsockname()
    s.close()
    return port

  @staticmethod
  def acquire_host_pair(host=None, port=None):
    if HAS_APP:
      host = host or app.get_options().tunnel_host
    assert host is not None, 'Must specify tunnel host!'
    port = port or TunnelHelper.get_random_port()
    return host, port

  @classmethod
  def wait_for_accept(cls, port, timeout=Amount(5, Time.SECONDS)):
    total_time = Amount(0, Time.SECONDS)
    sleep = cls.MIN_RETRY
    while total_time < timeout:
      try:
        accepted_socket = socket.create_connection(('localhost', port), timeout=5.0)
        accepted_socket.close()
        return True
      except socket.error:
        total_time += sleep
        time.sleep(sleep.as_(Time.SECONDS))
        sleep *= 2
    cls.log('timed out initializing tunnel')
    return False

  @classmethod
  def create_tunnel(cls, remote_host, remote_port, tunnel_host=None, tunnel_port=None):
    """
      Create or retrieve a memoized SSH tunnel to the remote host & port, using
      tunnel_host:tunnel_port as the tunneling server.
    """
    tunnel_key = (remote_host, remote_port)
    if tunnel_key in cls.TUNNELS:
      return 'localhost', cls.TUNNELS[tunnel_key][0]
    if HAS_APP:
      tunnel_host = tunnel_host or app.get_options().tunnel_host
    if tunnel_host is None:
      raise cls.TunnelError('Must specify tunnel host!')
    tunnel_port = tunnel_port or cls.get_random_port()
    cls.log('opening connection to %s:%s via %s:%s' %
        (remote_host, remote_port, tunnel_host, tunnel_port))
    ssh_cmd_args = ('ssh', '-T', '-L', '%d:%s:%s' % (tunnel_port, remote_host, remote_port),
                    tunnel_host)
    cls.TUNNELS[tunnel_key] = (tunnel_port, subprocess.Popen(ssh_cmd_args, stdin=subprocess.PIPE))
    if not cls.wait_for_accept(tunnel_port):
      raise cls.TunnelError('Could not establish tunnel to %s via %s' % (remote_host, tunnel_host))
    cls.log('session established')
    return 'localhost', tunnel_port

  @classmethod
  def create_proxy(cls, proxy_host=None, proxy_port=None):
    """
      Create or retrieve a memoized SOCKS proxy using the specified proxy host:port
    """
    if proxy_host in cls.PROXIES:
      return 'localhost', cls.PROXIES[proxy_host][0]
    proxy_host, proxy_port = cls.acquire_host_pair(proxy_host, proxy_port)
    cls.log('opening SOCKS proxy connection through %s:%s' % (proxy_host, proxy_port))
    ssh_cmd_args = ('ssh', '-T', '-D', str(proxy_port), proxy_host)
    cls.PROXIES[proxy_host] = (proxy_port, subprocess.Popen(ssh_cmd_args, stdin=subprocess.PIPE))
    if not cls.wait_for_accept(proxy_port):
      raise cls.TunnelError('Could not establish proxy via %s' % proxy_host)
    cls.log('session established')
    return 'localhost', proxy_port

  @classmethod
  def cancel_tunnel(cls, remote_host, remote_port):
    """<commit_msg>[PATCH 045/153] Give more time for SSH tunnels to establish in python twitter.common.<commit_after># ==================================================================================================
# Copyright 2012 Twitter, Inc.
# --------------------------------------------------------------------------------------------------
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this work except in compliance with the License.
# You may obtain a copy of the License in the LICENSE file, or at:
#
#  http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ==================================================================================================

import atexit
import errno
import socket
import subprocess
import time

from twitter.common import log
from twitter.common.quantity import Amount, Time

try:
  from twitter.common import app
  HAS_APP = True

  app.add_option(
    '--tunnel_host',
    type='string',
    dest='tunnel_host',
    default='nest1.corp.twitter.com',
    help='Host to tunnel commands through (default: %default)')

except ImportError:
  HAS_APP = False


__all__ = (
  'TunnelHelper',
)


def safe_kill(po):
  """
    Given a Popen object, safely kill it without an unexpected exception.
  """
  try:
    po.kill()
  except OSError as e:
    if e.errno != errno.ESRCH:
      raise
  po.wait()


# TODO(wickman) Add mox tests for this.
class TunnelHelper(object):
  """ Class to initiate an SSH or SOCKS tunnel to a remote host through a tunnel host.

  The ssh binary must be on the PATH.
  """
  TUNNELS = {}
  PROXIES = {}
  MIN_RETRY = Amount(5, Time.MILLISECONDS)
  DEFAULT_TIMEOUT = Amount(10, Time.SECONDS)

  class TunnelError(Exception): pass

  @classmethod
  def log(cls, msg):
    log.debug('%s: %s' % (cls.__name__, msg))

  @classmethod
  def get_random_port(cls):
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    s.bind(('localhost', 0))
    _, port = s.getsockname()
    s.close()
    return port

  @classmethod
  def acquire_host_pair(cls, host=None, port=None):
    if HAS_APP:
      host = host or app.get_options().tunnel_host
    assert host is not None, 'Must specify tunnel host!'
    port = port or cls.get_random_port()
    return host, port

  @classmethod
  def wait_for_accept(cls, port, tunnel_popen, timeout):
    total_time = Amount(0, Time.SECONDS)
    sleep = cls.MIN_RETRY
    while total_time < timeout and tunnel_popen.returncode is None:
      try:
        accepted_socket = socket.create_connection(('localhost', port), timeout=5.0)
        accepted_socket.close()
        return True
      except socket.error:
        total_time += sleep
        time.sleep(sleep.as_(Time.SECONDS))
        sleep *= 2
        tunnel_popen.poll()  # needed to update tunnel_popen.returncode
    if tunnel_popen.returncode is not None:
      cls.log('SSH returned prematurely with code %s' % str(tunnel_popen.returncode))
    else:
      cls.log('timed out initializing tunnel')
    return False

  @classmethod
  def create_tunnel(
        cls, remote_host, remote_port, tunnel_host=None, tunnel_port=None,
        timeout=DEFAULT_TIMEOUT):
    """
      Create or retrieve a memoized SSH tunnel to the remote host & port, using
      tunnel_host:tunnel_port as the tunneling server.
    """
    tunnel_key = (remote_host, remote_port)
    if tunnel_key in cls.TUNNELS:
      return 'localhost', cls.TUNNELS[tunnel_key][0]
    if HAS_APP:
      tunnel_host = tunnel_host or app.get_options().tunnel_host
    if tunnel_host is None:
      raise cls.TunnelError('Must specify tunnel host!')
    tunnel_port = tunnel_port or cls.get_random_port()
    cls.log('opening connection to %s:%s via %s:%s' %
        (remote_host, remote_port, tunnel_host, tunnel_port))
    ssh_cmd_args = ('ssh', '-T', '-L', '%d:%s:%s' % (tunnel_port, remote_host, remote_port),
                    tunnel_host)
    ssh_popen = subprocess.Popen(ssh_cmd_args, stdin=subprocess.PIPE)
    cls.TUNNELS[tunnel_key] = tunnel_port, ssh_popen
    if not cls.wait_for_accept(tunnel_port, ssh_popen, timeout):
      raise cls.TunnelError('Could not establish tunnel to %s via %s' % (remote_host, tunnel_host))
    cls.log('session established')
    return 'localhost', tunnel_port

  @classmethod
  def create_proxy(cls, proxy_host=None, proxy_port=None, timeout=DEFAULT_TIMEOUT):
    """
      Create or retrieve a memoized SOCKS proxy using the specified proxy host:port
    """
    if proxy_host in cls.PROXIES:
      return 'localhost', cls.PROXIES[proxy_host][0]
    proxy_host, proxy_port = cls.acquire_host_pair(proxy_host, proxy_port)
    cls.log('opening SOCKS proxy connection through %s:%s' % (proxy_host, proxy_port))
    ssh_cmd_args = ('ssh', '-T', '-D', str(proxy_port), proxy_host)
    ssh_popen = subprocess.Popen(ssh_cmd_args, stdin=subprocess.PIPE)
    cls.PROXIES[proxy_host] = (proxy_port, ssh_popen)
    if not cls.wait_for_accept(proxy_port, ssh_popen, timeout):
      raise cls.TunnelError('Could not establish proxy via %s' % proxy_host)
    cls.log('session established')
    return 'localhost', proxy_port

  @classmethod
  def cancel_tunnel(cls, remote_host, remote_port):
    """
{'fraction_non_alphanumeric': 0.15406427221172023, 'fraction_numerical': 0.004253308128544423, 'mean_word_length': 3.5025531914893615, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5923162', 'n_tokens_mistral': 3345, 'n_tokens_neox': 3251, 'n_words': 935}