Python em telecom em 7 minutos

Categoria: Python Publicado: Terça, 29 Maio 2012 Escrito por Helio Loureiro

Essa é uma apresentação que fiz na iMasters em abril último, no evento 7Masters sobre python

Não é um curso de python, nem nada próximo disso, mas apenas uma visão de que telecom é na verdade um grande TI, com aplicações que todos nós já conhecemos bem.

7_minutes_python_telecom.swf

 

Era uma apresentação que deveria levar 7 minutos.  Gastei 20.

UPDATE: 2021-12-23 troquei o arquivo embedded de flash pra um gif animado uma vez que nenhum browser sério ainda suporta swf.

 

UnicodeEncodeError: 'ascii' codec can't encode character

Categoria: Python Publicado: Sexta, 13 Abril 2012 Escrito por Helio Loureiro

Não é de hoje que me deparo com esses erros de unicode em python.

Em algumas funções que uso, principalmente as que lêem de timeline de redes sociais, tenho problemas com caracteres.  Dessa vez eu estava fazendo um programa que lê a timeline da rede social da empresa, que é baseada num produto lixo da Microsoft, o ShamePoint, digo, SharePoint, que publica a API de mensagens no formato RSS.  A função é pra gerar um gráfico de mensagens a cada 5 minutos para verificar o andamento da rede social (que passou do momento de hype).  Então como o ShamePoint é limitado em suas funções, resolvi fazer um "tracking" de posts através de uma função com um hash MD5 da mensagem postada.  

A idéia não é muito complexa, mas eis que fazendo o hash eu achei um:

helio@shibboleet:pynet$ ./mynet_tag_posts.py
Getting posts
Reading output and rss
Traceback (most recent call last):
  File "./mynet_tag_posts.py", line 106, in 
    main()
  File "./mynet_tag_posts.py", line 103, in main
    RSS = RSS_parser(XML)
  File "./mynet_tag_posts.py", line 60, in RSS_parser
    key = md5sum(rss.summary_detail.value)
  File "./mynet_tag_posts.py", line 19, in md5sum
    m.update(msg)
UnicodeEncodeError: 'ascii' codec can't encode character u'\xc5' in position 96: ordinal not in range(128)

E a função de hash MD5, a md5sum(), é super simples:

def md5sum(msg):
  m = hashlib.md5()
  m.update(msg)
  return m.hexdigest()

Só que mensagens com caracteres não ASCII, como em português ou suéco, simplesmente matam o processamento do MD5.

Em outros scripts eu contornei isso fazendo uma análise por caractere e substituindo todo aquele que tivesse valor maior que 128 por seu semelhante não acentuado na tabela ASCII.   Trabalho tosco, mas funcional.

Dessa vez resolvi procurar como corrigir isso.  E pra sempre.  Procurei na documentação do python sobre tratamento de unicode.

E não é que achei a solução do jeito mais simples e pythonista possível?  Basta forçar um atributo de formato de caractere no texto, como texto.encode(formato).  No caso, mudei a função para essa abaixo, usando formato de caracteres "utf-8":

def md5sum(msg):
  m = hashlib.md5()
  msg = msg.encode('utf-8')
  m.update(msg)
  return m.hexdigest()

E o problema foi resolvido.  Sem mais chororo de formatos iso-8859-1 e utf-8.  

Limpando os SPAMs do eri.cx

Categoria: Python Publicado: Sexta, 15 Abril 2011 Escrito por Helio Loureiro

Ultimamente, monitorando meu encurtador eri.cx, notei que o mesmo estava recebendo muitos links sem nenhum clique.  Olhando mais de perto pude ver que os mesmos eram relacionados à SPAM/vírus.  Alguns robôs de Internet, ou mesmo máquinas zumbis, que adicionam esses links suspeitos.

Um dos agressores foi claramente o IP 213.5.66.88.   Resolvi o problema adicionando-o em duas regras de firewall bem simples:

iptables -A INPUT -p tcp --dport 80 -s 213.5.66.88 -j LOG
iptables -A INPUT -p tcp --dport 80 -s 213.5.66.88 -j REJECT

Criei uma tabela chamada quarantine e migrei as entradas do banco para lá, manualmente.  E também notifiquei os administradores do site sobre o ocorrido, enviando mail para abuse@site e security@site.

Achei que a solução era perfeita e que tinha resolvido meus problemas.  Essa semana dei uma  olhada novamente na base do encurtador.  De 900 links inseridos, uns 350 eram desse tipo (malditos SPAMMERS!).

Chorei minhas mágoas no Twitter, bem no estilo "reclamar no Twitter muito", e recebi algumas sugestões para evitar esses ataques.  Mesmo meu guru mental para idéias técno-copiadas, Eduardo Maçan, que administra o encurtador miud.in -- do qual chupinhei várias coisas, sugeriu uma verificação do encurtador via web através do refereer antes de encurtar qualquer link.

Pensei em fazer isso, mas tornaria impossível notificar os administradores dos sites afetados que suas máquinas estão comprometidas, e fora o problema de performance que poderia ocorrer ainda mais que o hosting que contratei tem somente 64 MB de RAM e CPU compartilhada num ambiente Xen.  Então resolvi mudar a abordagem e simplesmente deixar as máquinas zumbis adicionarem as URLs "do mal".  Mas uma vez por dia eu rodo um programa em Python que faz a busca de padrões desses vírus/scammers, move pra base do quarentine (assim ninguém clicará nos mesmos) mantendo as mesmas entradas, e comunica ao mails abuse@domínio e security@domínio sobre o ataque.

Então segue o script que fiz para esse fim.

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

# COPYRIGHT
#
#    Copyright (c) 2011 by Helio A. L. Loureiro
#
#    Permission to use, copy, modify, distribute, and sell this software
#    for any purpose is hereby granted without fee, provided that the above
#    copyright notice appear in all copies and that both that copyright
#    notice and this permission notice appear in supporting documentation.
#    No representations are made about the suitability of this software for
#    any purpose. It is provided "as is" without express or implied
#    warranty.

### main code
from os import popen
from re import sub, search
import smtplib
from time import ctime,strftime


mysqldb = "eri.cx"
mysqluser = "eri.cx"
mysqlpasswd = "xc.ire"
log_dir = "/var/log"

# Enviando mensagens
def Log(msg):
	log_file = log_dir + "/report-sql-spam-" + \
		strftime("%Y-%m") + ".log"
	msg = msg.rstrip("\n")
	msg = "[ " + ctime() + " ] " + msg + "\n"
	log = open(log_file, "a")
	log.write(msg)
	log.close()

# Formatando o texto do mysql
def GenSQLCommand(sql):
	cmd = "mysql -u %s -p\'%s\' %s << EOF" % (mysqluser, mysqlpasswd, mysqldb)
	cmd = cmd + "\n" + sql + "\n" + "EOF\n"
	
	return cmd
	
def RetrieveMySQL():
	sql = """SELECT 
	yourls_url.keyword,
	yourls_url.url, 
	yourls_url.title,
	yourls_url.timestamp,
	yourls_url.ip,
	yourls_url.clicks
	FROM yourls_url WHERE  
	yourls_url.title like '%dingo.ucsf.edu%' OR
	yourls_url.title like '%cssa.grad.wustl.edu%' OR
	yourls_url.title like '%wiki.usfca.edu%' OR
	yourls_url.title like '%pantherfile.uwm.edu%' OR
	yourls_url.title like '%csci.marietta.edu%' OR
	yourls_url.title like '%webfiles.uci.edu%' OR
	yourls_url.title like '%Ultram %' OR
	yourls_url.title like '%Cialis%' OR
	yourls_url.title like '%Tramadol%' OR
	yourls_url.title like '%xanax%' OR
	yourls_url.title like '%viagra%' ORDER BY yourls_url.keyword;
	"""
	cmd = GenSQLCommand(sql)
	
	sql_result = popen(cmd).readlines()
	sql_result = sql_result[1:]

	return sql_result
	
def InsertQuarantine(db_lines):
	sql = """INSERT INTO yourls_quarantine 
	(keyword, url, title,timestamp,ip,clicks) VALUES """
	for e in db_lines:
		sql = sql + "\n( "
		e = e.rstrip("\n")
		for d in e.split("\t"):
			sql = sql + "\'" + d + "\', "
		sql = sql[:-2] + " ),"
	sql = sql[:-1]
	cmd = GenSQLCommand(sql)
	popen(cmd).read()
	
def DeleteEntries(keywords):
	sql = """DELETE FROM yourls_url USING yourls_url WHERE """ 
	sql = sql + "keyword = \"%s\"" % keywords[0]
	for k in range(1, len(keywords)):
		sql = sql + "OR keyword = \"%s\"" % keywords[k]
		
	cmd = GenSQLCommand(sql)
	popen(cmd).read()
	
def GetDomain(url):
	url = sub("^http://", "", url)
	url = sub("^https://", "", url)
	url = sub("/.*", "", url)
	
	return url

def SendMail(rcpts, msg):
	fromaddr = "helio" + "@" + "domain.com"
	mail = smtplib.SMTP('localhost')
	mail.sendmail(fromaddr, rcpts, msg)
	mail.quit()

def Main():
	spam_list = RetrieveMySQL()
	
	ATTACKED = {}
	KEYWORDS = []
	for spam in spam_list:
		Log(sub("\t", ",", spam))
		keyword, url, title, timestamp, ip, clicks = spam.split("\t")
		domain = GetDomain(url)
		KEYWORDS.append(keyword)
		try:
			ATTACKED[domain].append(spam)
		except KeyError:
			ATTACKED[domain] = [spam]
			
	for poor_soul in ATTACKED:
		rcpts = ["abuse@" + poor_soul, 
		"root@" + poor_soul,
		"security@" + poor_soul]
		msg = """From: Helio Loureiro <""" + "helio" + "@" + "domain.com" + """>
To: abuse@%s, root@%s, security@%s
Subject: Service abused from your domain

Hi,
		
I'm the sysadmin from domain eri.cx, a little poor shortener.
Recently I noticed some attacks (SPAM url shorteneds) coming from
your site.

Here follows the logs:

#timestamp|ip|title|url
""" % (poor_soul, poor_soul, poor_soul)
		for spam in ATTACKED[poor_soul]:
			keyword, url, title, timestamp, ip, clicks = spam.split("\t")
			msg = msg + "%s|%s|%s|%s\n" % (timestamp, ip, title, url)
		#print poor_soul, "=>", ATTACKED[poor_soul]
		
		msg = msg + """
It seems your server was compromised.  So please take actions accordingly.

This is an automatic mail report.  My apologizes if it sent unwanted 
information.

Best Regards,
Helio Loureiro
http://eri.cx
"""
		#print msg
		SendMail(rcpts, msg)
		InsertQuarantine(spam_list)
		DeleteEntries(KEYWORDS)

if __name__ == '__main__':
	Log("== Starting... == ")
	Main()
	Log("## Finishing... ## ")
	

 

 

Enviando IP por mail com Python

Categoria: Python Publicado: Sexta, 15 Abril 2011 Escrito por Helio Loureiro

De tanto @eduardomacan e @ale_borba reclamarem dos meus posts repetitivos no twitter, aqui segue então mais uma dica com Python pra dar um refresco na repetitividade.

Tenho algumas máquina velhas na empresa, umas Sun Ultra 10, que uso pra compilar e testar código.  Uma com Solaris 10 e outra com o 9 (ambas fazendo jus ao nome Slowlaris).  Elas tinham IP fixo e ficavam num ambiente de laboratório, mas recentemente resolveram tentar *cobrar* o espaço físico e a energia elétrica, além da "administração do sistema".   Resolvi então pegar ambas e as colocar próximas de mim, no ambiente de escritório.

O maior problema disso é que o ambiente disponibiliza IP somente por DHCP.  Então como resolver a questão de saber qual o IP da máquina para fazer conexão via SSH, uma vez que o mesmo pode mudar a cada novo boot (e realmente muda por aqui)?

Como sempre, Python veio para salvar o dia.  Uso o programa abaixo para enviar por mail qual o IP do servidor.  É feito para Solaris, por isso algumas definições são específicas, como o uso da interface "hme0" e a localização do comando "ifconfig".  Mas tudo pode ser adaptado facilmente para outra plataforma.

 

#! /usr/bin/python

import smtplib
from os import popen
from string import split
from re import search, sub
from socket import gethostname
from time import ctime

interface = "hme0"
ifconfig = "/usr/sbin/ifconfig"
mailserver = "mail.internal.domain.com"

def SendMail(rcpts, ip):
	fromaddr = "Este endereço de email está sendo protegido de spambots. Você precisa do JavaScript ativado para vê-lo."
	subject = "== Location update: %s (%s) == " % (ip, gethostname())
	msg = "From: %s\nTo: " % fromaddr
	for n in rcpts:
		msg = msg + n + ","
	msg = msg[:-1] + "\nSubject: %s" % subject
	msg = msg + """

Location update from host %s: %s

%s
-- Powered by Python --
""" % (gethostname(), ip, ctime())
	mail = smtplib.SMTP(mailserver)
	#mail.set_debuglevel(1)
	print "Sending:", msg
	mail.sendmail(fromaddr, rcpts, msg)
	mail.quit()
	
def GetIP():
	for line in popen(ifconfig + " " + interface).readlines():
		#print line
		if search("inet", line):
			line = line[:-1]
			line = sub(".*inet ", "", line)
			line = sub(" .*", "", line)
			
			return line
			
		
	
def Main():
	ip = GetIP()
	rcpts = [ "Este endereço de email está sendo protegido de spambots. Você precisa do JavaScript ativado para vê-lo.",
		"Este endereço de email está sendo protegido de spambots. Você precisa do JavaScript ativado para vê-lo.",
		"Este endereço de email está sendo protegido de spambots. Você precisa do JavaScript ativado para vê-lo."
		]
	SendMail(rcpts, ip)

if __name__ == '__main__':
	Main()

 

Em seguida apenas adicionei o mesmo dentro do "/etc/init.d" e fiz o link em "/etc/rc3.d".

Escrevendo uma aplicação com python-twitter (parte 3 - final)

Categoria: Python Publicado: Domingo, 03 Abril 2011 Escrito por Helio Loureiro

Após as partes 1 e 2 sobre como escrever uma aplicação com python-twitter, onde foi visto como instalar o pacote python-twitter ou via fonte ou via pacote Debian, e como registrar sua aplicação no Twitter, finalmente chegamos ao final: escrevendo a aplicação em python.

O objetivo de toda essa série de descrição de como utilizar uma aplicação de twitter-python era mostrar como fiz um script para automatizar o #FF, enviado às sexta-feiras.  

Inicialmente escrevi o seguinte script que fazia exatamente o #FF para todos os meus seguidores:

#! /usr/bin/python
import twitter

# App python-tweeter
# https://dev.twitter.com/apps/815176
api = twitter.Api(
    consumer_key = 'xxxxxxx',
    consumer_secret = 'yyyyyyyy',
    access_token_key = 'zzzzzzz',
    access_token_secret = 'zzzzzzzz'
    )
FOLLOWERS = []
users = api.GetFollowers()
for u in users:
        sc_name = str(u.screen_name)
        FOLLOWERS.append(['@' + sc_name])

SIZE = 140
MSG = "#FF"
i = 1
api.PostUpdate("Automated python-twitter-#FF mode=on")
for uid in FOLLOWERS:
    print i, uid
    if ( len(MSG + " " + uid) > SIZE):
        print MSG
        api.PostUpdate(MSG)
        sleep(60)
        MSG = "#FF " + uid
    else:
        MSG += " " + uid
    i += 1

print MSG
api.PostUpdate(MSG)

Esse pequeno script mostrou uma limitação do uso do python-twitter 0.8.1: a lista de usuários era buscada somente com os último 100 seguidores.  O demais não retornavam.  Eu corrigi esse comportamento e lançei o "0.8.2", mas na verdade ainda não foi aprovado.

Basicamente o módulo twitter, ou classe, permite buscar todos os seguidores com o método GetFollowers().  Cada seguidor retornado volta como uma matriz (array) de seguidores, com seus parâmetros como atributo do objeto retornado.  Então como exemplo um dos meus seguidores:

>>> print users[0]
{"description": "Expert in software development, photographer, open networker, 
 writer, handyman, old DJ+radiohost, idealist, creative, easily inspired, 
 very independent, spunky", 
"favourites_count": 44, 
"followers_count": 27800, 
"friends_count": 30580, 
"geo_enabled": true, 
"id": 5377742, 
"location": "Copenhagen, Denmark", 
"name": "J\u00f8rgen Larsen", 
"profile_background_color": "000000", 
"profile_background_tile": true, 
"profile_image_url": 
 "http://a0.twimg.com/profile_images/728474665/J_rgen_Larsen_org_nobackground_small_128x128_normal.png", 
"profile_link_color": "000000", 
"profile_sidebar_fill_color": 
 "http://a1.twimg.com/profile_background_images/78764489/background3.jpg", 
"profile_text_color": "333333", 
"protected": false, 
"screen_name": "porcupinex", 
"status": {"created_at": "Fri Apr 01 06:36:13 +0000 2011", 
    "favorited": false, 
    "id": 53707220474544128, 
    "source": "HootSuite", 
    "text": "RT @renejsum: RT @tinybuddha \"We must accept finite disappointment, but never lose infinite hope.\"
        ~Martin Luther King Jr. (Til @jesperdahl)", 
    "truncated": false}, 
"statuses_count": 1500, 
"time_zone": "Copenhagen", 
"url": "http://on.fb.me/porcupinex", 
"utc_offset": 3600}

chamando cada user[] como follower (for follower in user:), é possível verificar seus atributos usando algo como follower.url, follow.screen_name e assim por diante.  Em geral o que se vê no twitter é no formato @helioloureiro, que é uma referência ao atributo screen_name, esse sem o "@" originalmente.

Para publicar algo usando python-twitter, basta utilizar o método PostUpdate()

 

api.PostUpdate("Testando via twitter...")

Tudo isso já estaria de bom tamanho se não fossem duas coisas: o @ale_borba reclamou do flood de #FFs e realmente eu não queria mandar #FF para todo mundo, mas somente para quem realmente participa no Twitter.  Como eu tenho a política de seguir de volta quem me segue, tem muitos "robôs" que eu gostaria de evitar.  Então resolvi criar um grupo "amigos" e buscar somente os seguidores desse grupo:

 

Acabei escrevendo a busca de seguidores por grupo sem utlizar a API do python-twitter, com urllib2 e json:

 

lstname = 'Amigos'
AMIGOS = []
NOMES = {}
FOLLOWERS = {}
parameters = {}
parameters['cursor'] = -1
ic = 0
while (parameters['cursor'] != 0):
    url = '%s/%s/%s/members.json' % (api.base_url, user,lstname)
    
    json = api._FetchUrl(url, parameters=parameters)
    data = simplejson.loads(json)
    for u in data['users']:
        AMIGOS.append(u)
    parameters['cursor'] = int(data['next_cursor_str'])

    if (ic == 10): break
    ic += 1

Para evitar ainda um flood de mensagens (ou trollada), inclui ainda um randrange() pra criar uma aleatoriedade nos #FFs.  Comecei com 50%, mas aumentei pra 33% (1/3).  Também adicionei um belo sleep() para evitar o envio massivo de #FF, tornando um pouco mais suave as indicações.

O código final que utilizo atualmente é esse:

#! /usr/bin/python
import twitter
from time import sleep
from sys import exit
import simplejson
from random import randrange

# App python-tweeter
# https://dev.twitter.com/apps/815176
api = twitter.Api(
    consumer_key = 'xxxxxx',
    consumer_secret = 'yyyyyy',
    access_token_key = 'zzzzzzz',
    access_token_secret = 'kkkkkkkk'
    )

user = 'helioloureiro'
lstname = 'Amigos'
AMIGOS = []
NOMES = {}
FOLLOWERS = {}
parameters = {}
parameters['cursor'] = -1
ic = 0
while (parameters['cursor'] != 0):
    url = '%s/%s/%s/members.json' % (api.base_url, user,lstname)
    
    json = api._FetchUrl(url, parameters=parameters)
    data = simplejson.loads(json)
    for u in data['users']:
        AMIGOS.append(u)
    parameters['cursor'] = int(data['next_cursor_str'])

    if (ic == 10): break
    ic += 1

for a in AMIGOS:
    NOMES[a['screen_name']] = 1

users = api.GetFollowers()
for u in users:
    status = ""
    try:
        if (NOMES[u.screen_name] == 1):
            FOLLOWERS['@' + u.screen_name] = 1
            status = "OK"
    except:
        status = "NOT"

    print u.screen_name, status
    
SIZE = 140
MSG = "#FF"
i = 1
api.PostUpdates("Automated python-twitter-#FF mode=on")
for uid in FOLLOWERS.keys():
    if not (randrange(0,2)):
        print "Nao foi:", uid
        continue
    print i, uid
    if ( len(MSG + " " + uid) > SIZE):
        print MSG
        api.PostUpdate(MSG)
        sleep(60)
        MSG = "#FF " + uid
    else:
        MSG += " " + uid
    i += 1

sleep(60)
print MSG
api.PostUpdate(MSG)

api.PostUpdates("Python Twitter #rockz!!!")
api.PostUpdates("Automated python-twitter-#FF mode=off")

Com esse princípio de uso do python-twitter já criei várias aplicações, inclusive para re-enviar minhas publicações daqui e ainda utilizando o eri.cx para encurtar os link.

Mastodon Mastodon
We use cookies

We use cookies on our website. Some of them are essential for the operation of the site, while others help us to improve this site and the user experience (tracking cookies). You can decide for yourself whether you want to allow cookies or not. Please note that if you reject them, you may not be able to use all the functionalities of the site.