Tag Archives: python

SimpleHTTPServer with auth

import BaseHTTPServer
from SimpleHTTPServer import SimpleHTTPRequestHandler
import sys
import base64

key = ""

class AuthHandler(SimpleHTTPRequestHandler):
''' Main class to present webpages and authentication. '''
def do_HEAD(self):
print "send header"
self.send_response(200)
self.send_header('Content-type', 'text/html')
self.end_headers()

def do_AUTHHEAD(self):
print "send header"
self.send_response(401)
self.send_header('WWW-Authenticate', 'Basic realm=\"Test\"')
self.send_header('Content-type', 'text/html')
self.end_headers()

def do_GET(self):
global key
''' Present frontpage with user authentication. '''
if self.headers.getheader('Authorization') == None:
self.do_AUTHHEAD()
self.wfile.write('no auth header received')
pass
elif self.headers.getheader('Authorization') == 'Basic '+key:
SimpleHTTPRequestHandler.do_GET(self)
pass
else:
self.do_AUTHHEAD()
self.wfile.write(self.headers.getheader('Authorization'))
self.wfile.write('not authenticated')
pass

def test(HandlerClass = AuthHandler,
ServerClass = BaseHTTPServer.HTTPServer):
BaseHTTPServer.test(HandlerClass, ServerClass)

if __name__ == '__main__':
if len(sys.argv)<3: print "usage SimpleAuthServer.py [port] [username:password]" sys.exit() key = base64.b64encode(sys.argv[2]) test()

python scan IP range



import sh

print "Scanning..."
# ping range 8.8.8.7 - 8.8.8.9
for num in range(7,10):
    # declare ip address
    address = "8.8.8." + str(num)

    # check if host is alive using PING
    try:
        # bash equivalent: ping -c 1 > /dev/null
        sh.ping(address, "-c 1", _out="/dev/null")
        print "ping to", address, "OK"
    except sh.ErrorReturnCode_1:
        print "no response from", address


blocksync

#!/usr/bin/env python
"""
Synchronise block devices over the network

Copyright 2006-2008 Justin Azoff
Copyright 2011 Robert Coup
License: GPL

Getting started:

* Copy blocksync.py to the home directory on the remote host
* Make sure your remote user can either sudo or is root itself.
* Make sure your local user can ssh to the remote host
* Invoke:
sudo python blocksync.py /dev/source user@remotehost /dev/dest
"""

import sys
from sha import sha
import subprocess
import time

SAME = "same\n"
DIFF = "diff\n"

def do_open(f, mode):
f = open(f, mode)
f.seek(0, 2)
size = f.tell()
f.seek(0)
return f, size

def getblocks(f, blocksize):
while 1:
block = f.read(blocksize)
if not block:
break
yield block

def server(dev, blocksize):
print dev, blocksize
f, size = do_open(dev, 'r+')
print size
sys.stdout.flush()

for block in getblocks(f, blocksize):
print sha(block).hexdigest()
sys.stdout.flush()
res = sys.stdin.readline()
if res != SAME:
newblock = sys.stdin.read(blocksize)
f.seek(-len(newblock), 1)
f.write(newblock)

def sync(srcdev, dsthost, dstdev=None, blocksize=1024 * 1024):

if not dstdev:
dstdev = srcdev

print "Block size is %0.1f MB" % (float(blocksize) / (1024 * 1024))
# cmd = ['ssh', '-c', 'blowfish', dsthost, 'sudo', 'python', 'blocksync.py', 'server', dstdev, '-b', str(blocksize)]
cmd = ['ssh', '-c', 'blowfish', dsthost, 'python', 'blocksync.py', 'server', dstdev, '-b', str(blocksize)]
print "Running: %s" % " ".join(cmd)

p = subprocess.Popen(cmd, bufsize=0, stdin=subprocess.PIPE, stdout=subprocess.PIPE, close_fds=True)
p_in, p_out = p.stdin, p.stdout

line = p_out.readline()
p.poll()
if p.returncode is not None:
print "Error connecting to or invoking blocksync on the remote host!"
sys.exit(1)

a, b = line.split()
if a != dstdev:
print "Dest device (%s) doesn't match with the remote host (%s)!" % (dstdev, a)
sys.exit(1)
if int(b) != blocksize:
print "Source block size (%d) doesn't match with the remote host (%d)!" % (blocksize, int(b))
sys.exit(1)

try:
f, size = do_open(srcdev, 'r')
except Exception, e:
print "Error accessing source device! %s" % e
sys.exit(1)

line = p_out.readline()
p.poll()
if p.returncode is not None:
print "Error accessing device on remote host!"
sys.exit(1)
remote_size = int(line)
if size != remote_size:
print "Source device size (%d) doesn't match remote device size (%d)!" % (size, remote_size)
sys.exit(1)

same_blocks = diff_blocks = 0

print "Starting sync..."
t0 = time.time()
t_last = t0
size_blocks = size / blocksize
for i, l_block in enumerate(getblocks(f, blocksize)):
l_sum = sha(l_block).hexdigest()
r_sum = p_out.readline().strip()

if l_sum == r_sum:
p_in.write(SAME)
p_in.flush()
same_blocks += 1
else:
p_in.write(DIFF)
p_in.flush()
p_in.write(l_block)
p_in.flush()
diff_blocks += 1

t1 = time.time()
if t1 - t_last > 1 or (same_blocks + diff_blocks) >= size_blocks:
rate = (i + 1.0) * blocksize / (1024.0 * 1024.0) / (t1 - t0)
print "\rsame: %d, diff: %d, %d/%d, %5.1f MB/s" % (same_blocks, diff_blocks, same_blocks + diff_blocks, size_blocks, rate),
t_last = t1

print "\n\nCompleted in %d seconds" % (time.time() - t0)

return same_blocks, diff_blocks

if __name__ == "__main__":
from optparse import OptionParser
parser = OptionParser(usage="%prog [options] /dev/source user@remotehost [/dev/dest]")
parser.add_option("-b", "--blocksize", dest="blocksize", action="store", type="int", help="block size (bytes)", default=1024 * 1024)
(options, args) = parser.parse_args()

if len(args) < 2: parser.print_help() print __doc__ sys.exit(1) if args[0] == 'server': dstdev = args[1] server(dstdev, options.blocksize) else: srcdev = args[0] dsthost = args[1] if len(args) > 2:
dstdev = args[2]
else:
dstdev = None
sync(srcdev, dsthost, dstdev, options.blocksize)

python create crossword

import random, re, time, string
from copy import copy as duplicate

# optional, speeds up by a factor of 4
import psyco
psyco.full()

class Crossword(object):
def __init__(self, cols, rows, empty = '-', maxloops = 2000, available_words=[]):
self.cols = cols
self.rows = rows
self.empty = empty
self.maxloops = maxloops
self.available_words = available_words
self.randomize_word_list()
self.current_word_list = []
self.debug = 0
self.clear_grid()

def clear_grid(self): # initialize grid and fill with empty character
self.grid = []
for i in range(self.rows):
ea_row = []
for j in range(self.cols):
ea_row.append(self.empty)
self.grid.append(ea_row)

def randomize_word_list(self): # also resets words and sorts by length
temp_list = []
for word in self.available_words:
if isinstance(word, Word):
temp_list.append(Word(word.word, word.clue))
else:
temp_list.append(Word(word[0], word[1]))
random.shuffle(temp_list) # randomize word list
temp_list.sort(key=lambda i: len(i.word), reverse=True) # sort by length
self.available_words = temp_list

def compute_crossword(self, time_permitted = 1.00, spins=2):
time_permitted = float(time_permitted)

count = 0
copy = Crossword(self.cols, self.rows, self.empty, self.maxloops, self.available_words)

start_full = float(time.time())
while (float(time.time()) - start_full) < time_permitted or count == 0: # only run for x seconds self.debug += 1 copy.current_word_list = [] copy.clear_grid() copy.randomize_word_list() x = 0 while x < spins: # spins; 2 seems to be plenty for word in copy.available_words: if word not in copy.current_word_list: copy.fit_and_add(word) x += 1 #print copy.solution() #print len(copy.current_word_list), len(self.current_word_list), self.debug # buffer the best crossword by comparing placed words if len(copy.current_word_list) > len(self.current_word_list):
self.current_word_list = copy.current_word_list
self.grid = copy.grid
count += 1
return

def suggest_coord(self, word):
count = 0
coordlist = []
glc = -1
for given_letter in word.word: # cycle through letters in word
glc += 1
rowc = 0
for row in self.grid: # cycle through rows
rowc += 1
colc = 0
for cell in row: # cycle through letters in rows
colc += 1
if given_letter == cell: # check match letter in word to letters in row
try: # suggest vertical placement
if rowc - glc > 0: # make sure we're not suggesting a starting point off the grid
if ((rowc - glc) + word.length) <= self.rows: # make sure word doesn't go off of grid coordlist.append([colc, rowc - glc, 1, colc + (rowc - glc), 0]) except: pass try: # suggest horizontal placement if colc - glc > 0: # make sure we're not suggesting a starting point off the grid
if ((colc - glc) + word.length) <= self.cols: # make sure word doesn't go off of grid coordlist.append([colc - glc, rowc, 0, rowc + (colc - glc), 0]) except: pass # example: coordlist[0] = [col, row, vertical, col + row, score] #print word.word #print coordlist new_coordlist = self.sort_coordlist(coordlist, word) #print new_coordlist return new_coordlist def sort_coordlist(self, coordlist, word): # give each coordinate a score, then sort new_coordlist = [] for coord in coordlist: col, row, vertical = coord[0], coord[1], coord[2] coord[4] = self.check_fit_score(col, row, vertical, word) # checking scores if coord[4]: # 0 scores are filtered new_coordlist.append(coord) random.shuffle(new_coordlist) # randomize coord list; why not? new_coordlist.sort(key=lambda i: i[4], reverse=True) # put the best scores first return new_coordlist def fit_and_add(self, word): # doesn't really check fit except for the first word; otherwise just adds if score is good fit = False count = 0 coordlist = self.suggest_coord(word) while not fit and count < self.maxloops: if len(self.current_word_list) == 0: # this is the first word: the seed # top left seed of longest word yields best results (maybe override) vertical, col, row = random.randrange(0, 2), 1, 1 ''' # optional center seed method, slower and less keyword placement if vertical: col = int(round((self.cols + 1)/2, 0)) row = int(round((self.rows + 1)/2, 0)) - int(round((word.length + 1)/2, 0)) else: col = int(round((self.cols + 1)/2, 0)) - int(round((word.length + 1)/2, 0)) row = int(round((self.rows + 1)/2, 0)) # completely random seed method col = random.randrange(1, self.cols + 1) row = random.randrange(1, self.rows + 1) ''' if self.check_fit_score(col, row, vertical, word): fit = True self.set_word(col, row, vertical, word, force=True) else: # a subsquent words have scores calculated try: col, row, vertical = coordlist[count][0], coordlist[count][1], coordlist[count][2] except IndexError: return # no more cordinates, stop trying to fit if coordlist[count][4]: # already filtered these out, but double check fit = True self.set_word(col, row, vertical, word, force=True) count += 1 return def check_fit_score(self, col, row, vertical, word): ''' And return score (0 signifies no fit). 1 means a fit, 2+ means a cross. The more crosses the better. ''' if col < 1 or row < 1: return 0 count, score = 1, 1 # give score a standard value of 1, will override with 0 if collisions detected for letter in word.word: try: active_cell = self.get_cell(col, row) except IndexError: return 0 if active_cell == self.empty or active_cell == letter: pass else: return 0 if active_cell == letter: score += 1 if vertical: # check surroundings if active_cell != letter: # don't check surroundings if cross point if not self.check_if_cell_clear(col+1, row): # check right cell return 0 if not self.check_if_cell_clear(col-1, row): # check left cell return 0 if count == 1: # check top cell only on first letter if not self.check_if_cell_clear(col, row-1): return 0 if count == len(word.word): # check bottom cell only on last letter if not self.check_if_cell_clear(col, row+1): return 0 else: # else horizontal # check surroundings if active_cell != letter: # don't check surroundings if cross point if not self.check_if_cell_clear(col, row-1): # check top cell return 0 if not self.check_if_cell_clear(col, row+1): # check bottom cell return 0 if count == 1: # check left cell only on first letter if not self.check_if_cell_clear(col-1, row): return 0 if count == len(word.word): # check right cell only on last letter if not self.check_if_cell_clear(col+1, row): return 0 if vertical: # progress to next letter and position row += 1 else: # else horizontal col += 1 count += 1 return score def set_word(self, col, row, vertical, word, force=False): # also adds word to word list if force: word.col = col word.row = row word.vertical = vertical self.current_word_list.append(word) for letter in word.word: self.set_cell(col, row, letter) if vertical: row += 1 else: col += 1 return def set_cell(self, col, row, value): self.grid[row-1][col-1] = value def get_cell(self, col, row): return self.grid[row-1][col-1] def check_if_cell_clear(self, col, row): try: cell = self.get_cell(col, row) if cell == self.empty: return True except IndexError: pass return False def solution(self): # return solution grid outStr = "" for r in range(self.rows): for c in self.grid[r]: outStr += '%s ' % c outStr += '\n' return outStr def word_find(self): # return solution grid outStr = "" for r in range(self.rows): for c in self.grid[r]: if c == self.empty: outStr += '%s ' % string.lowercase[random.randint(0,len(string.lowercase)-1)] else: outStr += '%s ' % c outStr += '\n' return outStr def order_number_words(self): # orders words and applies numbering system to them self.current_word_list.sort(key=lambda i: (i.col + i.row)) count, icount = 1, 1 for word in self.current_word_list: word.number = count if icount < len(self.current_word_list): if word.col == self.current_word_list[icount].col and word.row == self.current_word_list[icount].row: pass else: count += 1 icount += 1 def display(self, order=True): # return (and order/number wordlist) the grid minus the words adding the numbers outStr = "" if order: self.order_number_words() copy = self for word in self.current_word_list: copy.set_cell(word.col, word.row, word.number) for r in range(copy.rows): for c in copy.grid[r]: outStr += '%s ' % c outStr += '\n' outStr = re.sub(r'[a-z]', ' ', outStr) return outStr def word_bank(self): outStr = '' temp_list = duplicate(self.current_word_list) random.shuffle(temp_list) # randomize word list for word in temp_list: outStr += '%s\n' % word.word return outStr def legend(self): # must order first outStr = '' for word in self.current_word_list: outStr += '%d. (%d,%d) %s: %s\n' % (word.number, word.col, word.row, word.down_across(), word.clue ) return outStr class Word(object): def __init__(self, word=None, clue=None): self.word = re.sub(r'\s', '', word.lower()) self.clue = clue self.length = len(self.word) # the below are set when placed on board self.row = None self.col = None self.vertical = None self.number = None def down_across(self): # return down or across if self.vertical: return 'down' else: return 'across' def __repr__(self): return self.word ### end class, start execution #start_full = float(time.time()) word_list = ['saffron', 'The dried, orange yellow plant used to as dye and as a cooking spice.'], \ ['pumpernickel', 'Dark, sour bread made from coarse ground rye.'], \ ['leaven', 'An agent, such as yeast, that cause batter or dough to rise..'], \ ['coda', 'Musical conclusion of a movement or composition.'], \ ['paladin', 'A heroic champion or paragon of chivalry.'], \ ['syncopation', 'Shifting the emphasis of a beat to the normally weak beat.'], \ ['albatross', 'A large bird of the ocean having a hooked beek and long, narrow wings.'], \ ['harp', 'Musical instrument with 46 or more open strings played by plucking.'], \ ['piston', 'A solid cylinder or disk that fits snugly in a larger cylinder and moves under pressure as in an engine.'], \ ['caramel', 'A smooth chery candy made from suger, butter, cream or milk with flavoring.'], \ ['coral', 'A rock-like deposit of organism skeletons that make up reefs.'], \ ['dawn', 'The time of each morning at which daylight begins.'], \ ['pitch', 'A resin derived from the sap of various pine trees.'], \ ['fjord', 'A long, narrow, deep inlet of the sea between steep slopes.'], \ ['lip', 'Either of two fleshy folds surrounding the mouth.'], \ ['lime', 'The egg-shaped citrus fruit having a green coloring and acidic juice.'], \ ['mist', 'A mass of fine water droplets in the air near or in contact with the ground.'], \ ['plague', 'A widespread affliction or calamity.'], \ ['yarn', 'A strand of twisted threads or a long elaborate narrative.'], \ ['snicker', 'A snide, slightly stifled laugh.'] a = Crossword(13, 13, '-', 5000, word_list) a.compute_crossword(2) print a.word_bank() print a.solution() print a.word_find() print a.display() print a.legend() print len(a.current_word_list), 'out of', len(word_list) print a.debug #end_full = float(time.time()) #print end_full - start_full

limit user bash shell with python

Simple use bdsh.py:

#!/usr/bin/python
# Copyright (C) 2013 - Remy van Elst

# 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 .

# This script can act as a shell for a user, allowing specific commands only.
# It tries its best to only allow those comamnds and strip possibly dangerous
# things like ; or >. But it won't protect you if you allow the vim command
# and the user executes !bash via vim (and such). It also logs everything to
# syslog for audit trailing purposes.

# It currently only checks commands, no parameters. This is on purpose.

import getpass, os, re, sys, syslog, signal, socket, readline

# format of whitelist: one command or regex per line
command_whitelist = "/etc/bdsh_whitelist.conf"
username = getpass.getuser()
hostname = socket.gethostname()

def log_command(command, status):
"""Log a command to syslog, either successfull or failed. """
global username
logline_failed = "[RESTRICTED SHELL]: user \"" + username + "\" NOT allowed for " + command
logline_danger = "[RESTRICTED SHELL]: user \"" + username + "\" dangerous characters in " + command
logline_success = "[RESTRICTED SHELL]: user \"" + username + "\" executed " + command
if status == "success":
syslog.syslog(logline_success)
elif status == "failed":
syslog.syslog(logline_failed)
elif status == "danger":
syslog.syslog(logline_danger)

def dangerous_characters_in_command(command):
# via http://www.slac.stanford.edu/slac/www/resource/how-to-use/cgi-rexx/cgi-esc.html
danger = [';', '&', '|', '>', '<', '*', '?', '`', '$', '(', ')', '{', '}', '[', ']', '!', '#'] for dangerous_char in danger: for command_char in command: if command_char == dangerous_char: return True def entire_command_scanner(command): danger = ["&&"] for dangerous_char in danger: if re.findall(dangerous_char, command): return True def execute_command(command): """First log, then execute a command""" log_command(command, "success") # try: # subprocess.call(command, shell=False) # except OSError: # pass os.system(command) def command_allowed(command, whitelist_file=command_whitelist): """Check if a command is allowed on the whitelist.""" try: with open(whitelist_file, mode="r") as whitelist: for line in whitelist: # We are reading commands from a file, therefore we also read the \n. if command + "\n" == line: return True else: continue except IOError as e: sys.exit("Error: %s" % e) def interactive_shell(): global username global hostname while True: prompt = username + "@" + hostname + ":" + os.getcwd() + " $ " try: if sys.version_info[0] == 2: command = raw_input(prompt) else: command = input(prompt) # Catch CRTL+D except EOFError: print("") sys.exit() if command == "exit" or command == "quit": sys.exit() elif command: if not entire_command_scanner(command): if command_allowed(command.split(" ", 1)[0]): for chars in command: if dangerous_characters_in_command(chars): log_command(command, "danger") # Don't let the user know via an interactive shell and don't exit command="" execute_command(command) if __name__ == "__main__": ## Catch CTRL+C / SIGINT. s = signal.signal(signal.SIGINT, signal.SIG_IGN) arguments = "" for args in sys.argv: if dangerous_characters_in_command(args): log_command(args, "danger") sys.exit() ## No Arguments? Then we start an interactive shell. if len(sys.argv) < 2: interactive_shell() else: ## Check if we are not launched via the local shell with a command (./shell.py ls) if sys.argv[1] and sys.argv[1] != "-c" and command_allowed(sys.argv[1].split(" ", 1)[0]) and not entire_command_scanner(sys.argv[1]): for arg in sys.argv[1:]: arguments += arg arguments += " " execute_command(arguments) ## Check if we are launched via the local shell and the command is not allowed elif len(sys.argv) < 3: for arg in sys.argv: arguments += arg arguments += " " log_command(arguments, "failed") elif sys.argv[2] and command_allowed(sys.argv[2].split(" ", 1)[0]) and not entire_command_scanner(sys.argv[2]): for arg in sys.argv[2:]: arguments += arg arguments += " " execute_command(arguments) else: for arg in sys.argv: arguments += arg arguments += " " log_command(arguments, "failed") # Debug use # print("\"" + arguments + "\"") ## Give back the CTRL+C / SIGINT signal.signal(signal.SIGINT, s)