pihunter.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
piHunter
Copyright (C) 2018  Renato Tavares <dr.renatotavares@gmail.com>

This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.

This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

You should have received a copy of the GNU General Public License
along with this program.  If not, see <http://www.gnu.org/licenses/>.
"""

import time
import socket
import argparse
import ipaddress
import threading

NUMBER_OF_THREADS = 100

q = Queue(100)  # Create the queue and thread pool
threads = []

def write_to_file(data, file_name):
    """Function to write data to a file.

    Just write the received data in a particular file on disk. Pre-formatting text 
    is the responsibility of the client.

    Args:
        data      (str): Data to be written.
        file_name (str): The file name.

    Returns:
        None
    """
    with open(file_name, "a+") as f:
       f.write(data)
       f.close()


def gen_ip_list(begin, end):
    """Function to generate a list of IPs.

    Generates an IP list within the requested range. Uses yield to generate on demand 
    each IP, allowing each thread to request an IP without competing with each other.

    Args:
        begin  (str): Initial IP
        end    (str): Final IP

    Yields:
        str: The next generated IP inside the range.

    Examples:
        Examples should be written in doctest format, and should illustrate how
        to use the function.

        >>> print([i for i in example_generator(4)])
        [0, 1, 2, 3]
    """

    begin, end = ipaddress.ip_address(begin), ipaddress.ip_address(end)
        
    while begin <= end:
        yield begin    
        begin += 1
    else
        return False



def check_open_port(ip='loalhost', port=22):
    """Function that checks if a door is open.

    Uses the socket module to check if a particular port is open. In this 
    release only IPv4 (AF_INET) can be verified through the TCP protocol (SOCK_STREAM)

    Args:
        ip   (str): IP to be checked.
        port (int): Port to be checked.

    Returns:
        bool: True for open port, False otherwise.
    """

    with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock:
        
        sock.settimeout(3)
        
        if sock.connect_ex((ip, port)) == 0:
            return True
        else:
            return False


def test_login(ip='localhost', user='pi', password='raspberry'):
    """Function to test if SSH login works.

    Uses the user and password passed to try to login via SSH. If the 
    login is successful then we will know the user and the password.

    Args:
        ip       (str): IP to be checked.
        user     (int): User to be tested on SSH.
        password (int): Password to be tested on SSH.

    Returns:
        bool: True for a successful login, False otherwise.
    """

    try:
        s = pxssh.pxssh()
        s.login(ip, user, password)
        # s.sendline('uptime')  # We can also execute multiple command s.sendline('uptime;df -h')
        # s.prompt()  # match the prompt
        # print(s.before)  # print everything before the prompt.
        s.logout()
        return True
    except:
        return False


def worker():
    
    while True:
        
        try:
            item = q.get()
        except Queue.Empty as e:
            # Handle empty queue here
            pass
        else:
            # Handle task here and call q.task_done()
            # Fazer seu trabalho
            q.task_done()


def start(initial_ip, final_ip):
    gen_ip_list(initial_ip, initial_ip)
    

    for i in range(NUMBER_OF_THREADS):
        t = threading.Thread(target=worker)
        t.start()
        threads.append(t)
    


if __name__ == "__main__":
    parser = argparse.ArgumentParser(prog='piHunter',
                                     description="%(prog)s Search for all the Raspberry Pi's with the default password in an IP ranges",
                                     epilog="Use this script for educational purposes only")  
    
    parser.add_argument("-v", "--verbose", help="enabling verbose mode", action="store_true", default=False)
    parser.add_argument("-o", "--output", help="path to the output file")

    parser.add_argument("begin", help="ip inicial")
    parser.add_argument("end", help="ip final")

    args = parser.parse_args()

    start(args.begin, args.end)