staging.inyokaproject.org

Python3 - socket

Status: Gelöst | Ubuntu-Version: Ubuntu 17.10 (Artful Aardvark)
Antworten |

Mokkujin

Anmeldungsdatum:
2. Mai 2008

Beiträge: 389

Hallo ihrs,

könnte mir mal bitte einer erklären warum socket.timeout nie anschlägt ? Bin gerade dabei mir python beizubringen und komme nicht auf die lösung.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
def checkPort(port,ip_addr,timeout):
	if _debug == 1:
		print ("\n--> check port {0} on ip {1}\n".format(port,ip_addr))
	try:
		_chk_port = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
		_chk_port.settimeout(timeout)
		_result = _chk_port.connect_ex((ip_addr, port))
		if _result == 0:
			_return = "Open"
		else:
			_return = "Closed"
		_chk_port.close()
	except KeyboardInterrupt:
		print ("You pressed Ctrl+C")
		sys.exit()
	except socket.gaierror:
		print ('Hostname could not be resolved. Exiting')
		sys.exit()
        except socket.timeout:
		_return = "timeout"
	except socket.error:
		print ("Couldn't connect to server")
		sys.exit()
	return _return

rklm Team-Icon

Projektleitung

Anmeldungsdatum:
16. Oktober 2011

Beiträge: 13242

Edit: zu früh

Marc_BlackJack_Rintsch Team-Icon

Ehemalige
Avatar von Marc_BlackJack_Rintsch

Anmeldungsdatum:
16. Juni 2006

Beiträge: 4735

@Mokkujin: Wo erwartest Du denn das eine socket.timeout-Ausnahme ausgelöst wird? connect_ex() wird das sicher nicht machen. Das liefert ja Fehlercodes statt Ausnahmen auszulösen. Versuch's mal ohne das _ex. ☺

Die führenden Unterstriche sind komisch. Ich nehme die immer für Variablen denen ich einen Namen geben muss, die aber nicht verwendet werden. So sehen das auch Werkzeuge wie Pylint. Wenn man ein Schlüsselwort der Sprache als Variablennamen nehmen möchte, ist die Konvention hinten einen Unterstrich anzuhängen. Also return_ statt _return. Allerdings ist das eh ein schlechter Name für ein ”Ding”. Wenn einem gar nichts anderes einfällt, dann kann man eigentlich immer result als Namen für das Ergebnis verwenden.

Der Name der Funktion hält sich nicht an den Style Guide for Python Code.

Die API bzw. da Verhalten ist auch komisch. Eine Funktion die etwas testet, sollte keine Ausgaben für den Benutzer machen, und schon gar nicht das ganze Programm beenden können und auf Strg+C reagieren.

Wo kommt _debug her? Konstanten schreibt man per Konvention KOMPLETT_IN_GROSSBUCHSTABEN. Es gibt auch ein praktisches logging-Modul in der Standardbibliothek.

Mokkujin

(Themenstarter)

Anmeldungsdatum:
2. Mai 2008

Beiträge: 389

Danke schon mal für die Antwort.

@Mokkujin: Wo erwartest Du denn das eine socket.timeout-Ausnahme ausgelöst wird? connect_ex() wird das sicher nicht machen. Das liefert ja Fehlercodes statt Ausnahmen auszulösen. Versuch's mal ohne das _ex. ☺

Wo du recht hast hast du recht *ARGL* ich frag jetzt einfach den return code nu fluppts

Könnte man das in Zeile 60 - 64 eleganter lösen ?

ich hab jetzt mal versucht deine Kritik umzusetzen, wenn dir noch etwas auffällt wäre es nett wenn du es mich wissen lässt

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
#!/usr/bin/env python3
# portscanner 
import socket, subprocess, sys, getopt
from datetime import datetime
# const
DEBUG = False
SHOW_ALL = False

def usage():
	print("portscanner.py")
	print("-" * 20)
	print("-h | --help			:  show this help")
	print("-n | --hostname=		:  define the hostname to be check")
	print("-s | --start_port=	   	:  define the start port")
	print("-e | --end_port=	      	:  define the end port")
	print("-a | --all-ports 		:  shows all (default is show only open ports)")
	print("-t | --timeout=			:  set the timeout for socket default is 3 seconds")
	print("-v | --verbose			:  set debug mode (set --all-ports)")

def get_ip_address(var_hostname):
	try:
		return socket.gethostbyname(var_hostname)
	except socket.error as err:
		print (str(err))
		sys.exit(4)

def check_port(var_port,var_ip,var_timeout):
	global DEBUG, SHOW_ALL
	if DEBUG == True:
		print ("\n--> check port {0} on ip {1}\n".format(var_port,var_ip))
	try:
		var_check_port = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
		var_check_port.settimeout(int(var_timeout))
		result = var_check_port.connect_ex((var_ip,var_port))
		if result == 0:
			return_ = "open"
		elif result == 111:
			return_ = "closed"
		elif result == 11:
			return_ = "firewalled"
		var_check_port.close()
	except KeyboardInterrupt: 
		sys.exit(1)
	except socket.gaierror:
		print ('Hostname could not be resolved. Exiting')
		sys.exit(3)
	except socket.error:
		print ("Couldn't connect to server")
		sys.exit(2)
	return return_

def scan_port():
	try:
		opts, args = getopt.getopt(sys.argv[1:], "h:n:s:e:at:v", ["help", "hostname=", "start_port=", "end_port=", "all-ports", "timeout=", "verbose"])	
	except getopt.GetoptError as err:
		print (str(err))
		usage()
		sys.exit(2)
	global DEBUG, SHOW_ALL	
	# var
	remote_host = ""
	var_str_port = ""
	var_end_port = ""
	var_socket_timeout = 3
	for opt, arg in opts:
		if opt in ("-h","--help"):
			usage()
			sys.exit(0)
		elif opt in ("-n","--hostname"):
			remote_host = arg
		elif opt in ("-s","--start_port"):
			var_str_port = int(arg)
		elif opt in ("-e","--end_port"):
			var_end_port = int(arg)
		elif opt in ("-a","--all-ports"):
			SHOW_ALL = True
		elif opt in ("-t","--timeout"):
			var_socket_timeout = int(arg)
		elif opt in ("-v","--verbose"):
			DEBUG = True
	# clear screen
	subprocess.call('clear', shell=True)
	# is debug on ?
	if DEBUG == True:
		SHOW_ALL = True
	# need more input ?
	if not remote_host:
		remote_host = input("please enter hostname or ip address : ")
	remote_host_ip = get_ip_address(remote_host)
	if not var_str_port:
		var_str_port = int(input("startport  : "))
	if not var_end_port:
		var_end_port = int(input("endport  : "))
	if var_str_port < 1 or var_str_port > 65535 or var_end_port < 1 or var_end_port > 65535:
		print("The Portrange must between 1-65535")
		sys.exit(1)

	
	# Print Banner
	print ("-" * 90)
	print ("please wait, scanning remote host {0}".format(remote_host_ip))
	print ("-" * 90)
	print ("\n")
	t1=datetime.now()
	for lp_check_port in range(var_str_port,var_end_port + 1):
		chk_result = check_port(lp_check_port,remote_host_ip,var_socket_timeout)
		if SHOW_ALL == False and chk_result == "Open":
			print ("Port {0}:     {1}".format(lp_check_port,chk_result))
		elif SHOW_ALL == True:
			print ("Port {0}:     {1}".format(lp_check_port,chk_result))
	t2 = datetime.now()
	time_take = t2 - t1
	print ("\n")
	print ("-" * 90)
	print ("scanning completed in : {0}".format(time_take))

if __name__ == "__main__":
	scan_port()

seahawk1986

Anmeldungsdatum:
27. Oktober 2006

Beiträge: 11278

Ich glaube da kommen eine Menge Angewohnheiten aus anderen Programmiersprachen durch... - es gibt sehr praktische Module in der Python STL wie argparse, die einem jede Menge Arbeit abnehmen können. Zwingend erforderliche Argumente würde ich als positionale statt als optionale Argumente umsetzen. Und man kann leicht ohne globale Variablen auskommen, wenn man die davon abhängigen Ausgaben in einer Funktion hält:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
#!/usr/bin/env python3
import argparse
import socket
import sys
from datetime import datetime

SOCKET_STATES = {
    0: "open",
    11: "firewalled",
    111: "closed",
}


def parse_args():
    parser = argparse.ArgumentParser()
    parser.add_argument("hostname", metavar="HOSTNAME",
                        help="define the hostname to be checked")
    parser.add_argument("start_port", type=int, metavar="START_PORT",
                        help="define the start port")
    parser.add_argument("end_port", type=int, metavar="END_PORT",
                        help="define the end port")
    parser.add_argument("-a", "--all-ports", action="store_true",
                        help="shows all (default is show only open ports)")
    parser.add_argument("-t", "--timeout", type=int, default=3,
                        help=("set the TIMEOUT for socket operations "
                              "(default is 3 seconds)"))
    parser.add_argument("-v", "--verbose", action="store_true",
                        help="set debug mode (set --all-ports)")
    args = parser.parse_args()
    if 0 < args.start_port <= 65535 and 0 < args.end_port <= 65535:
        return args
    else:
        sys.exit("The Portrange must between 1-65535")

def get_ip_address(hostname):
    try:
        return socket.gethostbyname(hostname)
    except socket.error as err:
        print (str(err))
        sys.exit(4)

def check_port(ip, port):
    try:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            result = s.connect_ex((ip, port))
            return SOCKET_STATES.get(result, result)
    except KeyboardInterrupt:
        sys.exit(1)
    except socket.gaierror:
        print ('Hostname could not be resolved. Exiting')
        sys.exit(3)
    except socket.error:
        print ("Couldn't connect to server")
        sys.exit(2)

def scan_ports(remote_host, start_port, end_port, debug=False, show_all=False):
    ip = get_ip_address(remote_host)

    # clear screen
    print(chr(27) + "[2J")
    # Print Banner
    print ("-" * 90)
    print ("please wait, scanning remote host {0}".format(ip))
    print ("-" * 90)
    print ("\n")

    t_start = datetime.now()
    for port in range(start_port, end_port + 1):
        if debug:
            print ("\n--> check port {0} on ip {1}\n".format(port, ip))
        port_state = check_port(ip, port)
        if port_state == "open" or show_all or debug:
            print ("Port {0: >5}: {1: >10}".format(port, port_state))
    duration = datetime.now() - t_start

    print ("\n")
    print ("-" * 90)
    print ("scanning completed in : {0}".format(duration))

if __name__ == '__main__':
    args = parse_args()
    socket.setdefaulttimeout(args.timeout)
    scan_ports(remote_host=args.hostname,
               start_port=args.start_port, end_port=args.end_port,
               debug=args.verbose, show_all=args.all_ports)

argparse generiert die Beschreibung der Parameter automatisch:

$ ./postscan.py 
usage: postscan.py [-h] [-a] [-t TIMEOUT] [-v] HOSTNAME START_PORT END_PORT
postscan.py: error: the following arguments are required: HOSTNAME, START_PORT, END_PORT
$ ./postscan.py -h
usage: postscan.py [-h] [-a] [-t TIMEOUT] [-v] HOSTNAME START_PORT END_PORT

positional arguments:
  HOSTNAME              define the hostname to be checked
  START_PORT            define the start port
  END_PORT              define the end port

optional arguments:
  -h, --help            show this help message and exit
  -a, --all-ports       shows all (default is show only open ports)
  -t TIMEOUT, --timeout TIMEOUT
                        set the TIMEOUT for socket operations (default is 3
                        seconds)
  -v, --verbose         set debug mode (set --all-ports) 

Mokkujin

(Themenstarter)

Anmeldungsdatum:
2. Mai 2008

Beiträge: 389

seahawk1986 schrieb:

Ich glaube da kommen eine Menge Angewohnheiten aus anderen Programmiersprachen durch... -

Da würde ich mal sagen schuldig im Sinne der Anklage 😉

es gibt sehr praktische Module in der Python STL wie argparse, die einem jede Menge Arbeit abnehmen können. Zwingend erforderliche Argumente würde ich als positionale statt als optionale Argumente umsetzen. Und man kann leicht ohne globale Variablen auskommen, wenn man die davon abhängigen Ausgaben in einer Funktion hält

Danke für den Tipp, ich versuch das mal umzusetzen und deinen Code 100% zu verstehen

Mokkujin

(Themenstarter)

Anmeldungsdatum:
2. Mai 2008

Beiträge: 389

So nu hätte ich ein zwei fragen dazu ...

was macht das denn nun wieder ? chr(27) = ESC ? aber was macht das "[2J" ?

1
 print(chr(27) + "[2J")

Versteh ich den Abschnitt hier richtig ?

1
return SOCKET_STATES.get(result, result)

schaut im Array SOCKET_STATES nach ob der return code drin ist wenn ja schickt er den value ?

seahawk1986

Anmeldungsdatum:
27. Oktober 2006

Beiträge: 11278

Mokkujin schrieb:

So nu hätte ich ein zwei fragen dazu ...

was macht das denn nun wieder ? chr(27) = ESC ? aber was macht das "[2J" ?

1
 print(chr(27) + "[2J")

Das ist eine ANSI-Sequenz für Clear Screen, man kann natürlich auch weiterhin clear mit subprocess aufrufen.

Versteh ich den Abschnitt hier richtig ?

1
return SOCKET_STATES.get(result, result)

schaut im Array SOCKET_STATES nach ob der return code drin ist wenn ja schickt er den value ?

Genau und falls nicht, schickt er den Wert von result.

Mokkujin

(Themenstarter)

Anmeldungsdatum:
2. Mai 2008

Beiträge: 389

chr(27) + "[2J" legt bei mir ein anderes Verhalten an den Tag als clear gibts da noch einen Trick ?

chr(27) + "[2J" mach den screen leer und fängt dann erst in der mitte an mit dem output

clear macht wirklich nur den screen leer und fängt in der ersten zeile an

seahawk1986

Anmeldungsdatum:
27. Oktober 2006

Beiträge: 11278

Eventuell reagiert dein Terminal-Emulator da nicht wie gehofft darauf (in ANSI_escape_code steht,dass nicht alle Geräte den Cursor dadurch in die obere linke Ecke setzen). Dann ist es doch besser clear aufzurufen.

Mokkujin

(Themenstarter)

Anmeldungsdatum:
2. Mai 2008

Beiträge: 389

Super danke dir ..

da du ja schon festgestellt hast das viel durch andere sprachen da reinspielt, wie könnte man das denn geschickter lösen ?

1
2
3
4
5
6
7
8
9
def write_log(logfile,entry,state):
	logging.basicConfig(format='%(asctime)s | %(levelname)s | %(message)s',filename=logfile,filemode='w',level=logging.DEBUG)
	print("Logfile {0} State {1} Eintrag {2}".format(logfile,state,entry))
	if state == "info":
		logging.info(str(entry))
	elif state == "warning":
		logging.warning(str(entry))
	elif state == "critical":
		logging.critical(str(entry))

seahawk1986

Anmeldungsdatum:
27. Oktober 2006

Beiträge: 11278

logging.basicConfig() hat laut Dokumentation nur einen Effekt, wenn der root-Logger noch keinen Handler hat - das macht also wenig Sinn das in die Funktion zu packen und immer wieder mit abweichenden Argumenten aufzurufen. Wenn du in unterschiedliche Dateien loggen willst, musst du dir dafür jeweils einen Log-Handler erstellen (vgl. https://docs.python.org/3/howto/logging.html#logging-advanced-tutorial)

Damit kann man sich die write_log() Funktion komplett sparen und einfach direkt logging.info, logging.warning usw. bzw. die die entsprechenden Methoden des jeweiligen Log-Handler aufrufen.

Falls es darum geht mit den Werten von state die gleichnamige Methode des Log-Handler aufzurufen, wäre getattr eine Möglichkeit - mal ganz grob ohne Fehlerbehandlung:

1
2
3
4
5
6
7
8
9
import logging

def write_log(msg, state):
    if state in ["info", "warning", "critical"]:
        log_method = getattr(logging, state)
        log_method(msg)

if __name__ == '__main__':
    write_log(msg="This is a warning", state="warning")

Man könnte auch ein Dictionary nutzen, wenn man von vornherein nur bestimmte Werte zulassen will:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
LOG_METHODS = {
"info": logging.info,
"warning": logging.warning,
"critical": logging.critical,
}

def write_log(msg, state):
    try:
        log_method = LOG_METHODS.get(state)
    except KeyError:
        logging.exception("invalid logging state %s", state)
    else:
        log_method(msg)

Mokkujin

(Themenstarter)

Anmeldungsdatum:
2. Mai 2008

Beiträge: 389

Also danke für deine Geduld ...

Dann werde ich das jetzt mal versuchen.

Marc_BlackJack_Rintsch Team-Icon

Ehemalige
Avatar von Marc_BlackJack_Rintsch

Anmeldungsdatum:
16. Juni 2006

Beiträge: 4735

My 2¢: Die Rückgabecodes ungleich Null in SOCKET_STATES hart als Zahlen rein zu schreiben ist nicht portabel. Dafür gibt es in errno Konstanten:

1
2
3
4
5
SOCKET_STATES = {
    0: 'open',
    errno.EAGAIN: 'firewalled',
    errno.ECONNREFUSED: 'closed',
}

In check_port() stehen immer noch Ausnahmebehandlungen die nie zum Zuge kommen. Einzig der socket.error könnte auftreten, dann aber nicht weil keine Verbindung zum Server aufgebaut werden könnte, sondern weil gar kein Socket-Objekt erstellt werden konnte. Und da weiss ich gar nicht ob das überhaupt jemals passieren kann im vorliegenden Szenario‽ Ausser viellecht wenn kein Speicher mehr verfügbar ist oder das Betriebssystem die maximale Anzahl von Sockets erstellt hat. In beiden Fällen ist die Ausgabe von "Couldn't connect to server" IMHO etwas zu allgemein und damit irreführend für den Benutzer des Programms.

Für den socket.gaierror müsste man alle Konstanten die zu dieser Ausnahme gehören in einem set() eintragen um später darauf prüfen zu können:

1
2
3
EAI_ERROR_CODES = set(
    getattr(socket, name) for name in dir(socket) if name.startswith('EAI_')
)

Und die Behandlung von KeyboardInterrupt ist in der Funktion fehl am Platz. So etwas gehört weiter oben in die Aufrufhierarchie. Damit bleibt dann das hier übrig:

1
2
3
4
5
6
7
def check_port(ip, port):
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        result = sock.connect_ex((ip, port))
        if result in EAI_ERROR_CODES:
            print('Hostname could not be resolved. Exiting')
            sys.exit(3)
        return SOCKET_STATES.get(result, result)

Allerdings verstehe ich nicht so ganz warum man hier connect_ex() verwendet und sich damit das auseineinanderpflücken der Fehlerwerte antut, statt Ausnahmen zu behandeln. Die wurden ja schliesslich erfunden, damit man sich nicht, oder zumindest nicht so intensiv, mit ”magischen” Fehlerwerten auseinandersetzen muss, und keine Fehler übersieht und nicht behandelt.

Die write_log()-Funktion macht keinen Sinn. Welches Problem soll diese Indirektion denn lösen?

Antworten |