OwlCyberSecurity - MANAGER
Edit File: test_connections.cpython-36.opt-1.pyc
3 ��JZ�P � @ s8 d Z ddlZddlZddlZddlmZ ddlmZ ddlmZ ddlmZ ddlm Z ddl Z ddl m Z dd l m Z dd l m Z dd l mZ dd l mZ dd l mZ ddl mZ ddl mZ ddlmZ ddlmZ ddlmZ ddlmZ ddlmZ ddlmZ ddlm Z ddlm Z ddlm Z ddlm Z ddlm Z ddlm!Z! dd lm"Z" dd lm#Z# dd lm$Z$ dd lm%Z% dd lm&Z& dd!lm'Z' dd"lm(Z( dd#lm)Z) dd$lm*Z* e j+� Z,G d%d&� d&e-�Z.G d'd(� d(e.e'j/�Z0G d)d*� d*e.e'j/�Z1G d+d,� d,e.e'j/�Z2G d-d.� d.e'j/�Z3e4d/k�r4e!e5� dS )0z;Tests for net_connections() and Process.connections() APIs.� N)�closing)�AF_INET)�AF_INET6)� SOCK_DGRAM)� SOCK_STREAM)�FREEBSD)�LINUX)�NETBSD)�OPENBSD)�OSX)�POSIX)�SUNOS)�WINDOWS)� supports_ipv6)�PY3)�AF_UNIX)� bind_socket)�bind_unix_socket)�check_connection_ntuple)�create_sockets)� get_free_port)�HAS_CONNECTIONS_UNIX)�pyrun)� reap_children)�run_test_module_by_name)� safe_rmpath)�skip_on_access_denied)�tcp_socketpair)�TESTFN)�TRAVIS)�unittest)�unix_socket_path)�unix_socketpair)� wait_for_filec @ s8 e Zd Zdd� Zdd� Zdd� Zd dd �Zdd d �ZdS )�Basec C s t stjdd�}d S )N�all)�kind)r �thisproc� connections)�self�cons� r+ �(/usr/lib64/python3.6/test_connections.py�setUp6 s z Base.setUpc C s" t t� t� ts tjdd�}d S )Nr% )r& )r r r r r' r( )r) r* r+ r+ r, �tearDown< s z Base.tearDownc C sr t jdd�}tdd� |D ��}tr.||j� S | jt|�d� |d jdkrf| j||j� j|j� � |d S d S )Nr% )r& c S s g | ]}|j |f�qS r+ )�fd)�.0�cr+ r+ r, � <listcomp>G s z+Base.get_conn_from_sock.<locals>.<listcomp>� r ���)r' r( �dictr �fileno� assertEqual�lenr/ )r) �sockr* Zsmapr+ r+ r, �get_conn_from_sockE s zBase.get_conn_from_sockNc C s� |dkr| j |�}t|� |jdkr6| j|j|j� � | j|j|j� | j|j|jtj tj �� |j � }| r�t r�t |t�r�|j� }|jtkr�|dd� }|jtkr�tr�n| j|j|� |jtkr�tr�tjdd�}| jtj� |� |S )z�Given a socket, makes sure it matches the one obtained via psutil. It assumes this process created one connection only (the one supposed to be checked). Nr3 � r% )r& r4 )r: r r/ r7 r6 �family�typeZ getsockopt�socketZ SOL_SOCKETZSO_TYPEZ getsocknamer � isinstance�bytes�decoder r r �laddrr r' r( �compare_procsys_connections�os�getpid)r) r9 �connrB r* r+ r+ r, � check_socketR s( zBase.check_socketr% c sd yt j|d�}W n t jk r0 tr*dS � Y nX � fdd�|D �}|j� |j� | j||� dS )z�Given a process PID and its list of connections compare those against system-wide connections retrieved via psutil.net_connections. )r& Nc s" g | ]}|j � kr|d d� �qS )Nr3 r4 )�pid)r0 r1 )rH r+ r, r2 � s z4Base.compare_procsys_connections.<locals>.<listcomp>)�psutil�net_connectionsZ AccessDeniedr �sortr7 )r) rH Z proc_consr& Zsys_consr+ )rH r, rC v s z Base.compare_procsys_connections)N)r% )�__name__� __module__� __qualname__r- r. r: rG rC r+ r+ r+ r, r$ 4 s $r$ c @ s| e Zd ZdZdd� Zeje� d�dd� �Zdd� Z eje� d�d d � �Z eje d �d d � �Z eje d �dd� �Z dS )�TestUnconnectedSocketsz;Tests sockets which are open but not connected to anything.c C sF dt � f}tttt|d��� }| j|�}| j|jtj � W d Q R X d S )Nz 127.0.0.1)�addr) r r r r r rG r7 �statusrI � CONN_LISTEN)r) rP r9 rF r+ r+ r, � test_tcp_v4� s z"TestUnconnectedSockets.test_tcp_v4zIPv6 not supportedc C sF dt � f}tttt|d��� }| j|�}| j|jtj � W d Q R X d S )Nz::1)rP ) r r r r r rG r7 rQ rI rR )r) rP r9 rF r+ r+ r, � test_tcp_v6� s z"TestUnconnectedSockets.test_tcp_v6c C sF dt � f}tttt|d��� }| j|�}| j|jtj � W d Q R X d S )Nz 127.0.0.1)rP ) r r r r r rG r7 rQ rI � CONN_NONE)r) rP r9 rF r+ r+ r, � test_udp_v4� s z"TestUnconnectedSockets.test_udp_v4c C sF dt � f}tttt|d��� }| j|�}| j|jtj � W d Q R X d S )Nz::1)rP ) r r r r r rG r7 rQ rI rU )r) rP r9 rF r+ r+ r, � test_udp_v6� s z"TestUnconnectedSockets.test_udp_v6z POSIX onlyc C sL t � �<}tt|td��� }| j|�}| j|jtj� W d Q R X W d Q R X d S )N)r= ) r! r r r rG r7 rQ rI rU )r) �namer9 rF r+ r+ r, � test_unix_tcp� s z$TestUnconnectedSockets.test_unix_tcpc C sL t � �<}tt|td��� }| j|�}| j|jtj� W d Q R X W d Q R X d S )N)r= ) r! r r r rG r7 rQ rI rU )r) rX r9 rF r+ r+ r, � test_unix_udp� s z$TestUnconnectedSockets.test_unix_udpN)rL rM rN �__doc__rS r �skipIfr rT rV rW r rY rZ r+ r+ r+ r, rO � s rO c @ sT e Zd ZdZejed�dd� �Zeje d�dd� �Z e e d�d d � �Z d d � Z d S )�TestConnectedSocketPairszJTest socket pairs which are are actually connected to each other. zunreliable on SUONSc C sz dt � f}tt|d�\}}zHtjdd�}| jt|�d� | j|d jtj � | j|d jtj � W d |j � |j � X d S )Nz 127.0.0.1)rP �tcp4)r& r; r r3 ) r r r r' r( r7 r8 rQ rI ZCONN_ESTABLISHED�close)r) rP �server�clientr* r+ r+ r, �test_tcp� s z!TestConnectedSocketPairs.test_tcpz POSIX onlyc C s2 t � �� }t|�\}}z�tjdd�}tr6dd� |D �}| jt|�d� tsRtsRt r�| j|d j d� | j|d j d� | j||d j p�|d j � nzt r�xt|d j |d j |d j |d j fD ]}| j|d� q�W n:| j|d j p�|d j |� | j|d j �p|d j |� W d |j � |j � X W d Q R X d S ) N�unix)r& c S s g | ]}|j d kr|�qS )z /var/run/log)�raddr)r0 r1 r+ r+ r, r2 � s z6TestConnectedSocketPairs.test_unix.<locals>.<listcomp>r; r � r3 )r! r" r' r( r r7 r8 r r r rd rB r r_ )r) rX r` ra r* rP r+ r+ r, � test_unix� s& "z"TestConnectedSocketPairs.test_unix)Zonly_ifc s � fdd�}t jd�}t jd�}ddlm} tjjt�}||�jt t �d|d�}||�jt t �d|d�}||�jt t �d |d�}||�jt t �d |d�} t |�} t t|��} t |�} t t|��} t� r�t |�}t t|��}t | �}t t|��}nd }d }d }d }� x�tj� D � ]�}|j� }� jt|�d � x�|D ]�}|j| jk�r\|||t t| f tjd� n�|j| jk�r�|||t t| f tjd� nZ|jt|dd �k�r�|||t t|f tjd� n,|jt|dd �k�r0|||t t|f tjd� �q0W �q W � j t |jdd� d S ) Nc s� d }t |� � j|j|� � j|j|� � j|j|� � j|j|� � j|j|� x"|D ]} | j| d �} | |krXqXqXW tr�� j | j |g� d S ) Nr% �inet�inet4�inet6�tcpr^ �tcp6�udp�udp4�udp6)r& ) r% rg rh ri rj r^ rk rl rm rn ) r r7 r< r= rB rd rQ r( r rC rH ) �procrF r< r= rB rd rQ ZkindsZ all_kindsr& r* )r) r+ r, � check_conn s z8TestConnectedSocketPairs.test_combos.<locals>.check_conna import socket, time s = socket.socket($family, socket.SOCK_STREAM) s.bind(('$addr', 0)) s.listen(1) with open('$testfn', 'w') as f: f.write(str(s.getsockname()[:2])) time.sleep(60) z� import socket, time s = socket.socket($family, socket.SOCK_DGRAM) s.bind(('$addr', 0)) with open('$testfn', 'w') as f: f.write(str(s.getsockname()[:2])) time.sleep(60) r )�Templatez 127.0.0.1)r< rP Ztestfnz::1r3 r% rg rh rj r^ rl rm rH ri rk rn z???)r& )r% rg rh rj r^ )r% rg rh rl rm )r% rg ri rj rk )r% rg ri rl rn ) �textwrap�dedent�stringrq rD �path�basenamer Z substitute�intr r r �evalr# r r' Zchildrenr( r7 r8 rH r rI rR r rU �getattr� assertRaises� ValueError)r) rp Z tcp_templateZ udp_templaterq ZtestfileZ tcp4_templateZ udp4_templateZ tcp6_templateZ udp6_templateZ tcp4_procZ tcp4_addrZ udp4_procZ udp4_addrZ tcp6_procZ tcp6_addrZ udp6_procZ udp6_addr�pr* rF r+ )r) r, � test_combos sb z$TestConnectedSocketPairs.test_combosc C s� t � ���}tjdd�}| jt|�t|�� tjdd�}| jt|�t� rJdnd� x,|D ]$}| j|jtt f� | j|j t � qVW tjdd�}| jt|�d� | j|d jt� | j|d j t � t� �rtjdd�}| jt|�d� | j|d jt � | j|d j t � tjd d�}| jt|�t� �r(dnd� x.|D ]&}| j|jtt f� | j|j t � �q4W tjd d�}| jt|�d� | j|d jt� | j|d j t � t� �r�tjd d�}| jt|�d� | j|d jt � | j|d j t � tjd d�}| jt|�t� �rd nd� x2|D ]*}| j|jtt f� | j|j t t f� �qW t� �r�tjdd�}| jt|�d� x.|D ]&}| j|jt � | j|j t t f� �qlW t �r�tjdd�}| jt|�d� x.|D ]&}| j|jt� | j|j t t f� �q�W W d Q R X d S )Nr% )r& rj r; r3 r^ r rk rl rm rn rg � ri rc � )r r' r( r7 r8 r �assertInr< r r r= r r r r )r) �socksr* rF r+ r+ r, � test_multi_sockets_filteringe s` z5TestConnectedSocketPairs.test_multi_sockets_filteringN)rL rM rN r[ r r\ r rb r rf r r r} r� r+ r+ r+ r, r] � s !br] c @ sJ e Zd ZdZe� dd� �Ze� dd� �Ze� eje o6e d�dd� ��Z d S ) �TestSystemWideConnectionsz Tests for net_connections().c s� � fdd�}t � �� ddlm} xZ|j� D ]N\}}|dkrBt rBq*|\}}tj|�}� jt|�tt |��� ||||� q*W � j t tjdd� W d Q R X d S )Nc sV t tdt� �}xB| D ]:}� j|j||d� |j|krF� j|j||d� t|� qW d S )Nr )�msg)ry r> �objectr� r< r= r )r* �families�types_r rF )r) r+ r, �check� s z0TestSystemWideConnections.test_it.<locals>.checkr )� conn_tmaprc z???)r& ) r �psutil._commonr� �itemsr rI rJ r7 r8 �setrz r{ )r) r� r� r& �groupsr� r� r* r+ )r) r, �test_it� s z!TestSystemWideConnections.test_itc C s@ t � �0}dd� tjdd�D �}| jt|�t|�� W d Q R X d S )Nc S s g | ]}|j tj� kr|�qS r+ )rH rD rE )r0 �xr+ r+ r, r2 � s z>TestSystemWideConnections.test_multi_socks.<locals>.<listcomp>r% )r& )r rI rJ r7 r8 )r) r� r* r+ r+ r, �test_multi_socks� s z*TestSystemWideConnections.test_multi_sockszunreliable on OSX + TRAVISc s t � �}t|�}W d Q R X g �d}xRt|�D ]F}tjjt�t|� }tj d| �}t |�}�j |j � | j t|� q,W x$t|�D ]}tt|� }t|� q�W �fdd�tjdd�D �}xJ�D ]B� | jt� fdd�|D ��|� tj� �} | jt| jd��|� q�W d S )N� a import time, os from psutil.tests import create_sockets with create_sockets(): with open('%s', 'w') as f: f.write(str(os.getpid())) time.sleep(60) c s g | ]}|j � kr|�qS r+ )rH )r0 r� )�pidsr+ r, r2 � s zFTestSystemWideConnections.test_multi_sockets_procs.<locals>.<listcomp>r% )r& c s g | ]}|j � kr|�qS r+ )rH )r0 r� )rH r+ r, r2 � s )r r8 �rangerD ru �realpathr �strrr rs r �appendrH Z addCleanupr r# rI rJ r7 �Processr( ) r) r� Zexpected�times�iZfname�srcZsprocZsysconsr| r+ )rH r� r, �test_multi_sockets_procs� s( z2TestSystemWideConnections.test_multi_sockets_procsN) rL rM rN r[ r r� r� r r\ r r r� r+ r+ r+ r, r� � s r� c @ s e Zd Zdd� ZdS )�TestMiscc C s� g }g }xXt t�D ]L}|jd�rtt|�}t|�}| jt|� | j||� |j|� |j|� qW trrtj tj t r|tj d S )NZCONN_) �dirrI � startswithry r� Z assertNotInr� r Z CONN_IDLEZ CONN_BOUNDr ZCONN_DELETE_TCB)r) ZintsZstrsrX ZnumZstr_r+ r+ r, �test_connection_constants� s z"TestMisc.test_connection_constantsN)rL rM rN r� r+ r+ r+ r, r� � s r� �__main__)6r[ rD r> rr � contextlibr r r r r rI r r r r r r r r r� r Zpsutil._compatr Z psutil.testsr r r r r r r r r r r r r r r r r! r" r# r� r' r� r$ ZTestCaserO r] r� r� rL �__file__r+ r+ r+ r, �<module> s^ \7 `P