python epoll

「イベント〜?」「ドリブ〜ン!」

なんか、すっかり放置していました
私は GUI でチャットなどを作るはずだったのです
しかし全く着手せぬまま現在に至っています、ごめんなさい

そこで、少しだけ epoll 触っとこうと思うことにして調べました
How To Use Linux epoll with Python
何とも丁寧な説明がありました

とりあえず

• select.epoll.register で file descriptor と何を監視するかを登録
• select.epoll.poll で active なやつをとってくる

ってのだけ把握

というわけで、少し試してみることに

import socket
import select
from OpenSSL.SSL import Connection, Context, SSLv23_METHOD

fqdn = 'www.google.com'

s = socket.socket()
s.settimeout(0)

epoll = select.epoll()
epoll.register(s.fileno(), 2**11-1)

print epoll.poll(1)
s.connect_ex((fqdn, 443))
print "connect"
print epoll.poll(1)
ctx = Context(SSLv23_METHOD)
print "make Context"
print epoll.poll(1)
c = Connection(ctx, s) 
print "make Connection"
print epoll.poll(1)
c.set_connect_state()
print "set connect state"
print epoll.poll(1)
c.do_handshake()
print "do handshake"
print epoll.poll(1)
x509 = c.get_peer_certificate()
print x509.get_subject()

で、結果がこちら

$ python test1.py 
[(3, 276)]
connect
[(3, 260)]
make Context
[(3, 260)]
make Connection
[(3, 260)]
set connect state
[(3, 260)]
Traceback (most recent call last):
  File "test1.py", line 26, in <module>
    c.do_handshake()
OpenSSL.SSL.WantReadError

non-blocking のまま handshake しようとして怒られました

select.EPOLLOUT = 4, select.EPOLLHUP = 16, select.EPOLLWRNORM = 256
みたいなんですけど、まぁ、OUT だけ見りゃいいのかな、この場合

教訓:「OpenSSL の handshake を呼ぶんなら blocking にしよう」
え、そこなの?

"OpenSSL.SSL.WantReadError"

暫く Twisted でねばってたんですが
SSL 周りを低いところから攻めてみようと思いまして
でもちょっとは楽したいなと思いながら

from twisted.internet import reactor
from twisted.internet.protocol import Protocol, ClientFactory
from twisted.internet.ssl import CertificateOptions

class MyProtocol(Protocol):
    def connectionMade(self):
        self.transport.socket.do_handshake()
        reactor.stop()

class MyProtocolFactory(ClientFactory):
    protocol = MyProtocol

if __name__ == '__main__':
    f = MyProtocolFactory()
    ctx = CertificateOptions()
    reactor.connectSSL('www.blogger.com', 443, f, ctx)
    reactor.run()

繋がったら切るってどういうことだよ!
っていうのを先ず書いてみたら

  File "test.py", line 7, in connectionMade
    self.transport.socket.do_handshake()
OpenSSL.SSL.WantReadError: 

例外ってしまいました

OpenSSL ってのは pyOpenSSL のことでしたが
まぁとりあえずググってみるかっつーダメな考えで
TLS connection with timeouts (and a few other difficulties) - Stack Overflow
とても丁寧な回答が投稿されてました

Twisted なのでそもそも reactor.connectSSL の socket は
timeout=30 ってのがデフォルト設定されていて
socket は non-blocking になっているので
handshake しに client hello して返事読もうとするんだけど
せっかちだからまだ返事来てなくってエラーになってるんで
落ち着いてちょっと待てってことでした
17.2.1 socket オブジェクト でも良く読めと

なので、待つことにしたら何も無しに終わるようになりました

@@ -1,10 +1,18 @@
+from time import sleep
+
 from twisted.internet import reactor
 from twisted.internet.protocol import Protocol, ClientFactory
 from twisted.internet.ssl import CertificateOptions
+from OpenSSL.SSL import WantReadError
 
 class MyProtocol(Protocol):
     def connectionMade(self):
-        self.transport.socket.do_handshake()
+        while True:
+            try:
+                self.transport.socket.do_handshake()
+                break
+            except WantReadError:
+                sleep(0.1)
         reactor.stop()
 
 class MyProtocolFactory(ClientFactory):

なんか、ちょっと可哀そうだから sleep 入れてみた
入れないと 2000 回以上 WantReadError してました
コンピューターは速いなぁ

あ、ほんとは、
Protocol とかでなくて twisted.protocols.basic.LineReceiver とかで
line を送ったり受けとったりし始めたらいいんです
handshake なんて裏でちょちょいとやってくれます
でもほら、
Protocol と ClientFactory だけで色々試してみたいじゃないですか

virtualenv activate

pyOpenSSL に実装されてる機能の数が残念だというお話は
今日も明日もググったー: 16進数 python
でさせて頂いてあったわけですが
以前に証明書の署名アルゴリズムを string で表示させる method と
Connection からサーバの提示する証明書のリストを list で返す method と
見様見真似で実装してみて投稿してみたんですけど
「メソッド足して欲しかったらテスト書けよテスト」
って言われて
「えぇ〜テストなんて書いたことないし、プログラマじゃないしぃ〜」
と自暴自棄になっていた時期もありました

で、テスト書こうということであれば真っ新な環境の方がいいなぁと思い
そういうのって virtualenv とか使うんだって試したことあったんですが
既に忘却の彼方だったので復習する為に検索することに
でまぁ、やることっつったら install して env 作って activate するだけみたい
仕組みだったら
virtualenv - def __mopemope__(self, *args, **kwargs):
で解説されてるし!
あっ、っていうか
モダンPython開発環境入門 - def __mopemope__(self, *args, **kwargs):
こっちの方が興味ありですし!

で、pip? って思ったんですが easy_install (setuptools?) の次らしいです
Why I like pip かを語ってらっしゃる方もいらっしゃることだし
「最近人気の、easy_install よりも高機能なインストールコマンド。」と言われたら
とりあえず試してみたくなるのが人情というもの
とりあえずだから bootstrap とか言わずに

cd /tmp
wget http://www.python.org/ftp/python/2.6.2/Python-2.6.2.tar.bz2
tar xfj Python-2.6.2.tar.bz2
cd Python-2.6.2
./configure --prefix=/tmp/local
make -j 4
make install

cd /tmp
wget http://pypi.python.org/packages/source/v/virtualenv/virtualenv-1.3.3.tar.gz
tar xfz virtualenv-1.3.3.tar.gz
/tmp/local/bin/python virtualenv-1.3.3 sandbox
source sandbox/bin/activate

wget http://pypi.python.org/packages/source/p/pip/pip-0.4.tar.gz
python pip-0.4/pip.py install ipython

とりあえず ipython 入っちゃった

pyOpenSSL のテストを試してみないとなぁと思ったので

cd /tmp
wget http://jaist.dl.sourceforge.net/sourceforge/pyopenssl/pyOpenSSL-0.9.tar.gz
tar xfz pyOpenSSL-0.9.tar.gz
cd pyOpenSSL-0.9
python setup.py build
python setup.py install

と install してみて

$ cd test
$ python test_ssl.py
............
----------------------------------------------------------------------
Ran 12 tests in 1.009s

OK
$

テスト通ったー!! ってまだ何もしてないですけどね

これで patch あてたもんを install してテスト書いてみました
get_signature_algorithm 向けには用意されてる証明書の署名アルゴリズムで確認
sha1WithRSAEncryption で署名されてました

    def test_get_signature_algorithm(self):
        cert = load_certificate(FILETYPE_PEM, self.pemData)
        self.assertEqual(cert.get_signature_algorithm(), "sha1WithRSAEncryption")

get_peer_cert_chain 向けには verisign.com に繋いで

• get_peer_cert_chain の結果の最初が get_peer_certificate の結果と合うか
• get_peer_cert_chain の結果のそれぞれの CN が想定されるものか

を確認
EV SGC ですねぇ

    def test_get_peer_cert_chain(self):
        context = Context(SSLv3_METHOD)
        client = socket()
        client.connect(('verisign.com', 443))
        clientSSL = Connection(context, client)
        clientSSL.set_connect_state()
        clientSSL.do_handshake()
        cert = clientSSL.get_peer_certificate()
        certs = clientSSL.get_peer_cert_chain()
        self.assertEqual(dump_certificate(FILETYPE_PEM, cert),
                dump_certificate(FILETYPE_PEM, certs[0]))
        self.assertEqual(certs[0].get_subject().CN, 'www.verisign.com')
        self.assertEqual(certs[1].get_subject().CN,
                'VeriSign Class 3 Extended Validation SSL SGC CA')
        self.assertEqual(certs[2].get_subject().CN,
                'VeriSign Class 3 Public Primary Certification Authority - G5')

他の test でも verisign.com に繋いでるんで無駄な気もしつつ
でも test って分けた方がいいの? と思いつつ
こういう場合はどうしたらいいんだろうなぁ

やぁ、virtualenv 素敵だなぁ

16進数 python

Python の OpenSSL のバインディング? っていうと
OpenSSLの本にも載っている M2Crypto
でもこれ SWIG を入れなきゃいけなかったりしてちょっとインストールが面倒
他にも SWIG 使ってる package 使いたいってんならいいんだけど、そうでもないし
SWIG 使ってる何か良いパッケージあるんでしょうか

で、簡単にインストールできるのが pyOpenSSL
これは easy_install でサクっと入ってくれたりするんだけど非力
何が例えば困ったかっていうと

• X.509 の署名アルゴリズムが取れない
• SSL で繋いだときのサーバが提示するサーバ証明書以外の証明書が取れない
• 公開鍵のパラメータが取れない

とかまぁ、ニッチな要望なんですけど

で、patch でも書くかと思っていじっていたんですが
RSA っていうと大きな素数を使います、2 つ
しかも掛け算して 1024bit とか 2048bit でないと表現できないような数字を使うので
OpenSSL は BIGNUM とかって多倍長? って言うんでしょうか、型が用意されていて
その型で計算しています、桁溢れちゃうからね!!
で、pyOpenSSL はどうも BIGNUM をバインドしてくれていない、ひどい

ひどい! と思ったんですがそこで諦めてはいかんと思い
かといって BIGNUM を実装というのは踏み込んではいけない世界な気がして
OpenSSL: Documents, bn(3) を良く眺めてみると

char *BN_bn2hex(const BIGNUM *a);

とある、これは BIGNUM を char で 16 進表示してくれるようでした
string までもってったら Python の int にもできるだろうと思って検索
2-36進数の文字列をint型に変換する - スコトプリゴニエフスク通信にありました

>>> int("0101", 2)
5
>>> int("nishio", 36)
1422284208

おぉ〜

で、これを C で書かないといけないので include/python2.6/intobject.h をチェック

PyAPI_FUNC(PyObject *) PyInt_FromString(char*, char**, int);

名前からして何かこれっぽいけど、何だこの第二引数は、ということで更に "PyInt_FromString" で検索
7.2.1 (通常)整数型オブジェクト (plain integer object)
素晴しい、日本語だ
で、どうも NULL で良いらしいので NULL にしたら無事大きな数字が出てきました

NULL ってのはアドレス 0 を挿すポインタのことらしい
要するに 0?

結局どうなったかっていうと

static PyObject *
crypto_PKey_params(crypto_PKeyObj *self, PyObject *args)
{

PyObject *params;

if (!PyArg_ParseTuple(args, ":params"))
    return NULL;

params = PyDict_New();

if (self->pkey->type == EVP_PKEY_RSA) {
  if (self->pkey->pkey.rsa->n != NULL)
    PyDict_SetItem(params, PyString_FromString("n"), PyInt_FromString(BN_bn2hex(self->pkey->pkey.rsa->n), NULL, 16));
  if (self->pkey->pkey.rsa->e != NULL)
    PyDict_SetItem(params, PyString_FromString("e"), PyInt_FromString(BN_bn2hex(self->pkey->pkey.rsa->e), NULL, 16));
  if (self->pkey->pkey.rsa->d != 0)
    PyDict_SetItem(params, PyString_FromString("d"), PyInt_FromString(BN_bn2hex(self->pkey->pkey.rsa->d), NULL, 16));
  if (self->pkey->pkey.rsa->p != NULL)
    PyDict_SetItem(params, PyString_FromString("p"), PyInt_FromString(BN_bn2hex(self->pkey->pkey.rsa->p), NULL, 16));
  if (self->pkey->pkey.rsa->q != NULL)
    PyDict_SetItem(params, PyString_FromString("q"), PyInt_FromString(BN_bn2hex(self->pkey->pkey.rsa->q), NULL, 16));
  if (self->pkey->pkey.rsa->dmp1 != NULL)
    PyDict_SetItem(params, PyString_FromString("dmp1"), PyInt_FromString(BN_bn2hex(self->pkey->pkey.rsa->dmp1), NULL, 16))
;
  if (self->pkey->pkey.rsa->dmq1 != NULL)
    PyDict_SetItem(params, PyString_FromString("dmq1"), PyInt_FromString(BN_bn2hex(self->pkey->pkey.rsa->dmq1), NULL, 16))
;
  if (self->pkey->pkey.rsa->iqmp != NULL)
    PyDict_SetItem(params, PyString_FromString("iqmp"), PyInt_FromString(BN_bn2hex(self->pkey->pkey.rsa->iqmp), NULL, 16))
;
} else if (self->pkey->type == EVP_PKEY_DSA) {
  if (self->pkey->pkey.dsa->p != NULL)
    PyDict_SetItem(params, PyString_FromString("p"), PyInt_FromString(BN_bn2hex(self->pkey->pkey.dsa->p), NULL, 16));
  if (self->pkey->pkey.dsa->q != NULL)
    PyDict_SetItem(params, PyString_FromString("q"), PyInt_FromString(BN_bn2hex(self->pkey->pkey.dsa->q), NULL, 16));
  if (self->pkey->pkey.dsa->g != 0)
    PyDict_SetItem(params, PyString_FromString("g"), PyInt_FromString(BN_bn2hex(self->pkey->pkey.dsa->g), NULL, 16));
  if (self->pkey->pkey.dsa->priv_key != NULL)
    PyDict_SetItem(params, PyString_FromString("priv_key"), PyInt_FromString(BN_bn2hex(self->pkey->pkey.dsa->priv_key), NU
LL, 16));
  if (self->pkey->pkey.dsa->pub_key != NULL)
    PyDict_SetItem(params, PyString_FromString("pub_key"), PyInt_FromString(BN_bn2hex(self->pkey->pkey.dsa->pub_key), NULL
, 16));
}

return params;
}

こんなんになっちゃったんだけど、何か格好悪くない?
誰かに直して欲しい、あと、unittest の書き方も教えて欲しい
つか先ず BIGNUM をバインドしてほしい

あ、ちなみに、Ruby の source を中々参考にしました。