Python – Demonologia

artwork by ~laurD on DeviantART

Python jest świetnym językiem do całej palety zastosowań. Służy on pomocą zarówno administratorom systemów, osobom które potrzebują napisać prosty parser, a nawet twórcom aplikacji webowych – dzięki takim frameworkom jak Django. W poniższym tekscie chciałbym pokazać jak łatwo jest stworzyć usługę systemową (ang. daemon) w języku Python, która będzie wykonywała dla nas określoną czynność – jako przykład przedstawię fakedns – bardzo prostego daemona DNS, który na dowolne zapytanie o adres IP będzie zwracał zawsze ten sam.

Dlaczego Python? Co z innymi językami?

Nie lepiej mieć aplikację którą można skompilować? Absolwenci kierunków informatycznych jeszcze kilka lat temu dowiadywali się na studiach, że najbardziej zaawansowanym językiem do wszelakich zastosowań jest Java. Mieli praktyczne doświadczenie z tworzenia aplikacji w Javie… i zaczęli pisać. Kod który chciałbym tu przedstawić pisałem dwa lata temu. Zanim się za niego zabrałem, zacząłem od szukania już gotowego rozwiązania za pomocą Google. Znalazłem, nawet kilka. Jedno w Javie. O zgrozo. Ja wiem, że się da. Wiem, że Java jest naprawde zaawansowanym językiem. Gotowa aplikacja wymaga jednak dużego nadkładu. Trzeba przygotować środowisko, przygotować konkretną ilość pamięci… i czekać kiedy aplikacja się wyłoży. Takie mam przynajmniej doświadczenia z aplikacjami w Javie, które pisali programiści zaraz po studiach, pracując w profesjonalnej firmie software’owej.

A inne języki? Natualnie. Jest C. Świetny język, ale czy mamy wystarczająco dużo czasu aby w nim pisać nawet małe aplikacje? .NET ? Wolne żarty. Nie dość, że pod Linuxem jest bardzo grubymi nićmi przyszywany, to i nie mamy pewności, czy Microsoft nie powie pewnego dnia hola, jedynie na ich, komercyjnej platformie. Naddatek potrzebnej mocy obliczeniowej, pamięci, wymaganej przestrzeni dyskowej… i ta świadomość, że nie jest to dopracowane i pewnie nigdy nie będzie – niczym Samba próbująca naśladować również kalekie i dziurawe, Microsoftowe domeny.

A co ze skryptowymi? Mnie na myśl przychodzą trzy alternatywy – Lua, Ruby i skrypt w bashu.

Lua – niesamowicie przejrzysty język. Piekielnie szybki, realizowany przez wręcz mikroskopijną bibliotekę. Na jego niekorzyść trzeba jednak zaliczyć wręcz niemal konieczność liczenia się z faktem, że jeśli rozwiązanie które potrzebujemy stworzyć, będzie wymagało biblioteki – będziemy zmuszeni długo jej szukać lub przygotować samemu jej obsługę za pomocą języka C.

Ruby – chyba najbliższy Pythonowi. Dlaczego nie. Bibliotek jest coraz więcej, jest wydajny. Miałem okazję jakiś czas temu nawet coś napisać w tym języku (a dokładnie – przepisać mały framework z Pythona), ale… jakoś w Pythonie pisze mi się lepiej. Co ważne – Ruby jest głęboko zakorzeniony w Perlu. Programistom mówi się jak powinni pisać kod, jednak zawsze mogą robić w to w stylu Perlowym – pisać szybko i wydajnie, jednak przeczytanie i przeanalizowanie go po jakimś czasie staje się piekłem. Kod w Pythonie jest de facto zawsze taki sam, nie zależnie kto by go pisał – czytanie go jest proste i przyjemne.

No i jeszcze uzupełnie o język skryptowy – Bash. Można w nim naprawdę dużo zrobić. Warto jednak zastanowić się na ile aplikacja będzie zaawansowana i czy zależy nam na wydajności. Podam przykład. Stworzyłem kiedyś skrypt w bashu generujący w oparciu o kilka plików wejściowych reguły dla firewalla, reguły shapera oraz treść dhcpd.conf. Skrypt wewnątrz wywoływał po kilka tysięcy razy awk/sed’a itp. Na maszynie jednoprocesorowej (Pentium 4 1.4 GHz), która pełniła funkcję routera dla sieci 100 Mbit/s, przy mniejszym obciążeniu – skrypt wykonywał się kilka minut. Przy większym – godzinę. Za każdym razem uruchamiając awk, spawnował odrębnego basha. Cóż. Przepisałem rozwiązanie na język Lua. Skrypt niezależnie od obciążenia maszyny wykonywał się w 0.02 sekundy. Jeśli byłby to Python czy Ruby – efekt byłby zapewne zbliżony do Lua.

Tworzenie usługi systemowej dla Linux/Unix

OK, zaczynamy. Usługa systemowa powinna przede wszystkim:
- wiedzieć jak pisać do sysloga
- zrzucić uprawnienia root’a
- odkleić się od stdin i stdout i przejść w tło
Jeśli taka usługa dodatkowo potrzebuje obsługiwać jakiś systemowy port TCP/UDP (o numerze niższym niż 1024) – zanim zrzuci ona prawa roota, powinna otworzyć sobie określonego socketa.

Nie będę tłumaczył poszczególnych fragmentu kodu ponieważ uważam, że jest on wystarczająco przejżysty.

fakedns.py

#!/usr/bin/python -u
# -*- coding: utf-8 -*-
" Fake DNS server: Always answer with the same, defined as conf.destip reply, listen at conf.listen "

class conf:
    listen='ADRES_IP_NA_KTÓRYM_APLIKACJA_MA_NASŁUCHIWAĆ'
    destip='ADRES_IP_NA_KTÓRY_APLIKACJA_BĘDZIE_KIEROWAĆ'

import os,sys,pwd,grp
import logging,logging.handlers
import socket

log=logging.getLogger('fakedns')
logging.basicConfig()
logging.root.setLevel(level=logging.INFO)
#logging.root.setLevel(level=logging.DEBUG)
hdlr=logging.handlers.SysLogHandler('/dev/log')
log.addHandler(hdlr)

def drop_privileges(uid_name='nobody',gid_name='nogroup'):
    starting_uid=os.getuid()
    starting_gid=os.getgid()
    starting_uid_name=pwd.getpwuid(starting_uid)[0]
    log.info('fakedns: drop_privileges: started as %s/%s'%(pwd.getpwuid(starting_uid)[0],grp.getgrgid(starting_gid)[0]))
    if os.getuid()!=0:
        log.info("fakedns: drop_privileges: already running as '%s'"%starting_uid_name)
        return
    if starting_uid==0:
        # Get the uid/gid from the name
        running_uid = pwd.getpwnam(uid_name)[2]
        running_gid = grp.getgrnam(gid_name)[2]
        # Try setting the new uid/gid
        try:
            os.setgid(running_gid)
        except OSError,e:
            log.error('Could not set effective group id: %s'%e)
        try:
            os.setuid(running_uid)
        except OSError,e:
            log.error('Could not set effective user id: %s'%e)
        # Ensure a very convervative umask
        new_umask=077
        old_umask=os.umask(new_umask)
        log.info('fakedns: drop_privileges: Old umask: %s, new umask: %s'%(oct(old_umask),oct(new_umask)))
    final_uid=os.getuid()
    final_gid=os.getgid()
    log.info('fakedns: drop_privileges: running as %s/%s'%(pwd.getpwuid(final_uid)[0],grp.getgrgid(final_gid)[0]))

def become_daemon(home='.',stdin='/dev/null', stdout='/dev/null', stderr='/dev/null'):
    log.info('fakedns: deamonize')
    # perform first fork
    try:
        if os.fork()>0:
            sys.exit(0)
    except OSError,e:
        log.info("fakedns: 1st fork failed: %s"%e.strerror)
        sys.exit(1)
    os.setsid()
    os.chdir(home)
    os.umask(0)
    # second fork
    try:
        if os.fork()>0:
            sys.exit(0)
    except OSError,e:
        log.info("fakedns: 2nd fork failed: %s"%e.strerror)
        sys.exit(1)
    i=open(stdin,'r')
    o=open(stdout,'a+')
    e=open(stderr,'a+',0)
    os.dup2(i.fileno(),sys.stdin.fileno())
    os.dup2(o.fileno(),sys.stdout.fileno())
    os.dup2(e.fileno(),sys.stderr.fileno())

class DNSQuery:
    def __init__(self, data):
        self.data=data
        self.dominio=''

        tipo = (ord(data[2]) >> 3) & 15     # Opcode bits
        if tipo == 0: # Standard query
            ini=12
            lon=ord(data[ini])
            while lon != 0:
                self.dominio+=data[ini+1:ini+lon+1]+'.'
                ini+=lon+1
                lon=ord(data[ini])

    def respuesta(self, ip):
        packet=''
        if self.dominio:
            packet+=self.data[:2] + "\x81\x80"
            # Questions and Answers Counts
            packet+=self.data[4:6] + self.data[4:6] + '\x00\x00\x00\x00'
            # Original Domain Name Question
            packet+=self.data[12:]
            # Pointer to domain name
            packet+='\xc0\x0c'
            # Response type, ttl and resource data length -> 4 bytes
            packet+='\x00\x01\x00\x01\x00\x00\x00\x3c\x00\x04'
            # 4bytes of IP
            packet+=str.join('',map(lambda x: chr(int(x)), ip.split('.')))
        return packet

if __name__ == '__main__':
    udps = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    try:
        udps.bind((conf.listen,53))
    except socket.error, (val,e):
        log.info("fakedns: %s"%e)
        sys.exit(1)
    log.info('fakedns: dom.query. 60 IN A %s' % conf.destip)
    drop_privileges()
    become_daemon()
    try:
        while 1:
            data, addr = udps.recvfrom(1024)
            #log.info(addr)
            p=DNSQuery(data)
            udps.sendto(p.respuesta(conf.destip), addr)
            #log.info('%s -> %s' % (p.dominio, conf.destip))
            log.info('fakedns: %s -> %s'%(p.dominio,addr[0]))
    except KeyboardInterrupt:
        log.info('fakedns: done.' % (p.dominio, conf.destip))
        udps.close()
    except:
        log.info('fakedns: crashed serving %s (%s)'%(addr, data))

Powiadomienia SMS mBank mobile – drugie podejście

Dwa miesiące temu przedstawiłem dwa sposoby w jaki można sobie uruchomić powiadomienia SMS od aplikacji lub o przychodzących mailach. Oba miały pewne wady więc zdecydowałem się powrócić do tematu raz jeszcze.

Pierwszym rozwiązaniem, było wykorzystanie +48xxxxxxxxx@text.plusgsm.pl
Rozwiązanie dobre, bo dostajemy natychmiast powiadomienie na komórkę. Dodatkowo większość skrzynek pocztowych ma możliwość ustawienia automatycznej kopii na drugi adres – więc w dosłownie chwilę możemy sobie ustawić automatyczne powiadomienie o mailach.
Rozwiązanie to ma jednak swoją wadę – dostajemy pełną treść, która nie zmieści się w jednym SMS’ie jeśli w mailu jest jakakolwiek treść poza samym tematem.

Drugim zasugerowanym przeze mnie rozwiązaniem było wykorzystanie możliwości wysyłania powiadomień SMS z Google Calendar. To rozwiązanie również jest dobre. Możemy wykorzystać konto w Google Apps, dzięki czemu adres powiadomień jest bardziej uniwersalny, ale ma też swoje wady – chodzi mianowicie o dzienny limit powiadomień SMS z kalendarza Google. Jak się przekonałem, taki limit jest. Dokładnej liczby nie znam, ale w przybliżeniu to około 25 SMS’ów.

Rozwiązanie optymalne?

Postanowiłem zatem poszukać czegoś po środku. Wykorzystałem obsługę skrzynki Google do pobierania wiadomości, a następnie jedynie nadawcę i temat przekazuje do usługi text.plusgsm.pl. Wydaje mi się, że rozwiązanie jest optymalne. Oto kod:

#!/usr/bin/env python
# -- encoding: utf8 --

__author__ = 'mw AT nme.pl'
__version__ = '1.4'
__date__ = 'pią, 25 wrz 2009, 13:22:28 CEST'

class setup:
    verbose = False
    class google:
        login = 'KONTO@GOOGLE'
        password = 'HASŁO'
    class notify:
        recipient="+48xxxxxxxxx@text.plusgsm.pl"

import poplib
import email
import email.Header
#import sms
import smtplib
import re

class Fwd:

    def __init__(self,address,subject):

        mesg = "From: %s\nTo: %s\nSubject: %s\n\n" % (address, setup.notify.recipient, subject)
        server = smtplib.SMTP('smsb.plusgsm.pl')
        if setup.verbose:
            server.set_debuglevel(1)
        server.sendmail (address, setup.notify.recipient, mesg)
        server.quit()

class Mailbox:

    def __init__(self):

        self.server = poplib.POP3_SSL('pop.gmail.com', 995)
        self.server.user(setup.google.login)
        self.server.pass_(setup.google.password)
        self.server.set_debuglevel(0)

        # fetch
        count = self.server.stat()[0]
        if setup.verbose:
                print count
        if count:
            for i in range(1,count+1):
                (header, msg, octets) = self.server.retr (i)
                mail = email.message_from_string('\n'.join(msg))
                address = email.Header.decode_header(mail['From'])[1]
                if address[1]:
                    address = address[0].decode(address[1],'ignore')
                else:
                    address = address[0]
                address = re.sub('>\s*$','',re.sub('^\s*< ','',address))
                subject = email.Header.decode_header(mail['Subject'])[0]
                if subject[1]:
                    subject = subject[0].decode(subject[1],'ignore')
                else:
                    subject = subject[0]
                #print subject
                #sms.parse(subject.split())
                Fwd(address,subject)
                #self.server.dele (i)

        # commit
        self.server.quit()

mbox = Mailbox()

Aby te powiadomienia działały, na koncie shellowym umieszczamy skrypt, nadajemy mu prawa do uruchomienia, przechodzimy do edycji crona:

chmod +x pop2smsV2.py
crontab -e

dodajemy zadanie do crona, aby był uruchamiany co minutę:

* * * * * /home/NAZWA/bin/pop2smsV2.py >/dev/null 2>&1

Podsumowanie

Teraz to rozwiązanie najbardziej przypomina mi to które miałem w Erze. Trzeba niestety użyć do tego shella z cronem. OVH odpada z dwóch powodów – brak crona oraz brak dostępu do sieci z poziomu shella. Może i by się dało uruchomić to jako CGI z ich webowym cronem, ale niestety mija się to trochę z celem, ponieważ cron OVH może być uruchamiany najczęściej co godzinę.

Przewagą tego rozwiązania nad powiadomieniem z Ery jest brak limitu skrzynki 12 MB. Limitem jest ponad 7 GB co nie stanowi już problemu. Jeśli nie chcemy zostawiać na skrzynce śladu – możemy usunąć komentarz z linii self.server.dele (i) – co spowoduje kasowanie przesłanych jako SMS wiadomości.

Zaletą obsługi powiadomień za pomocą crona jest to, że możemy ustalić sztywne godziny w których skrypt ma działać – możemy ustawić, że powiadomienia mogą przychodzić pomiędzy 8:00, a 16:00 oraz 18:00 – 22:00 (trzeba się wtedy liczyć z faktem, że o ósmej rano otrzymamy wszystkie SMS’y z nocy). Jeśli chcemy mieć powiadomienia aktywne w godzinach 8:30 – 22:30, robimy to następującymi regułami:

30-59   8 * * * /home/nme/sms-notify/pop2smsV2.py >/dev/null 2>&1
*    9-21 * * * /home/nme/sms-notify/pop2smsV2.py >/dev/null 2>&1
0-30   22 * * * /home/nme/sms-notify/pop2smsV2.py >/dev/null 2>&1

Dodatkowo – dzięki temu rozwiązaniu mamy adres do powiadomień, znacznie łatwiejszy do zapamiętania – bo w postaci np. powiadomienia@nasza-domena.pl

Podgląd strony do pliku jpg?

Poruszenie w Internecie tematyką wprowadzania cenzury, o której ostatnio pisałem, podsunęło mi pewien pomysł, który mam nadzieję niebawem Wam przedstawię. W między czasie uznałem, że w projekcie przyda mi się funkcjonalność zautomatyzowanego robienia podglądów (thumbnail) stron WWW do obrazków. Miałem przy tym całkiem dużo frajdy i mam nadzieję, że efekty mojej pracy komuś się to przydadzą.

Serwisów które przygotowują podgląd stron jest w sieci niewiele, a takich, które za darmo udostępniają wygodne API, nie mają śmiesznie małych limitów przygotowań podglądów nie znalazłem. Napisałem sobie takie narzędzie sam. Jeśli ktoś chce wykorzystać ten kod – proszę bardzo – proszę tylko o informację gdzie jest on dostępny oraz liczę na darmowy dostęp

Twoja własna mini-przeglądarka

Jak szybko zrobić sobie przeglądarkę z obsługą Flasha, JavaScriptu i tych wszystkich urozmaiceń? Wystarczyło mi kilkanaście minut Zdecydowałem się na użycie GTK (domyślnego zestawu kontrolek dla m.in. Gnome) oraz Webkit’a – silnika do renderowania wykorzystywanego w takich przeglądarkach jak Safari (Mac OS), Google Chrome (Windows) czy Midori (Linux):

#!/usr/bin/env python

url = 'http://www.nme.pl'
size = [800, 600]

import gtk,webkit
window = gtk.Window(gtk.WINDOW_TOPLEVEL)
window.set_default_size(size[0], size[1])
window.connect('destroy', gtk.main_quit)
preview = webkit.WebView()
preview.open(url)
scrolled = gtk.ScrolledWindow()
scrolled.add(preview)
window.add(scrolled)
window.show_all()
gtk.main()

Właściwy podgląd strony

Kod trzeba było trochę rozbudować. Konfiguracja jest w klasie conf. Uruchamiając kod w pierwszym parametrze należy podać adres strony (nie zapominając o nazwie protokołu, np. http://www.nme.pl).

w katalogu w którym znajduje się skrypt trzeba założyć podkatalog thumb, w którym zapisywane będą obrazki jpg.

shot.py:

#!/usr/bin/env python

__author__ = 'mw AT nme.pl'
__version__ = '1.0'
__date__ = 'wto, 25 sie 2009, 09:15:48 CEST'

import pygtk
pygtk.require('2.0')
import gtk,gtk.gdk,gobject
import webkit
import hashlib
import os,sys

class conf:
    show_scrollbars = True
    show_progress = True
    script_timeout_secs = 15
    size = [1024, 768]

class Snapshot:

    def load_progress_changed(self, view, progress):
        print '%d%%' % progress

    def load_finished(self, view, frame):
        print 'loading finished'
        w = gtk.gdk.get_default_root_window()
        wp = self.window.get_position()
        ws = self.window.get_size()
        sz = w.get_size()
        pb = gtk.gdk.Pixbuf(gtk.gdk.COLORSPACE_RGB,False,8,ws[0],ws[1])
        pb = pb.get_from_drawable(w,w.get_colormap(),wp[0],wp[1],0,0,ws[0],ws[1])
        if (pb != None):
            pb.save('thumbs/'+self.file+'.jpg','jpeg',{'quality':'95'})
        gtk.main_quit()

    def __init__(self, url=None):

        self.url = url
        self.file = hashlib.sha512(url).hexdigest()
        self.overtime = False
        self.window = gtk.Window(gtk.WINDOW_TOPLEVEL)
        self.window.set_default_size(conf.size[0], conf.size[1])
        self.window.set_decorated(False)
        self.window.connect('destroy', gtk.main_quit)
        self.preview = webkit.WebView()
        self.preview.open(self.url)
        if conf.show_progress:
            self.preview.connect('load-progress-changed',self.load_progress_changed)
        self.preview.connect('load-finished',self.load_finished)
        self.scrolled = gtk.ScrolledWindow()
        if conf.show_scrollbars:
            self.scrolled.props.hscrollbar_policy = gtk.POLICY_AUTOMATIC
            self.scrolled.props.vscrollbar_policy = gtk.POLICY_AUTOMATIC
        else:
            self.scrolled.props.hscrollbar_policy = gtk.POLICY_NEVER
            self.scrolled.props.vscrollbar_policy = gtk.POLICY_NEVER
        self.scrolled.add(self.preview)
        self.window.add(self.scrolled)
        self.window.show_all()
        gobject.timeout_add(conf.script_timeout_secs*1000, self.timeout)
        gtk.main()

    def timeout(self):
        gtk.main_quit()
        sys.exit(1)

if len(sys.argv)==2:
    Snapshot(url=sys.argv[1])
else:
    print >> sys.stderr, "Usage: %s url\n" % sys.argv[0]
    sys.exit(1)

Wrapper do generowania właściwego poglądu

Skrypt działa, ale wymaga działania X’ów, coś nam jeszcze wyświetla i zapisuje obrazek w skali 1:1. Da się to zrobić w tle. Potrzebujemy do tego dwóch narzędzi – convert z pakietu imagemagick oraz pakietu xvfb – Virtual Framebuffer ‘fake’ X server. Mając już te narzędzia – całą robotę zrobi za nas skrypt o nazwie snap:

#!/bin/bash

state=0
plik=`python -c "import hashlib; print hashlib.sha512('$1').hexdigest()+'.jpg'"`
[ -e "thumbs/$plik" ] || {
	xvfb-run -a --server-args='-screen 0 1024x768x16' python shot.py $1
}
state=$?
[ "$state" -eq "0" ] && {
	[ -e "thumbs/$plik" ] && {
		convert -scale 310x -quality 75 thumbs/$plik thumbs/$plik
	} || {
		exit 1
	}
} || {
	exit 1
}

skryptowi nadajemy prawa do jego uruchamiania:

chmod +x snap

i uruchamiamy:

./snap http://www.nme.pl

w katalogu thumb odnajdujemy gotowego jpg’a (jego nazwa jest hashem sha512 URL’a). Wygląda to tak:

Powiązane, teksty w języku angielskim:

1. Podobne rozwiązanie w oparciu o Qt, warto obejrzeć.
2. Przykładowy opis jak można uruchomić serwis do generowania podglądów w oparciu o Django z nawiązaniem do gotowych już rozwiązań.