aboutsummaryrefslogtreecommitdiffstats
path: root/protocols/skype/test.py
blob: 233c41c8542ca47215008847639fad933ee73a34 (plain)
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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
#!/usr/bin/env python2.7

import subprocess
import sys
import pexpect
import unittest
import shutil
import os
import hashlib

def openssl(args):
	with open(os.devnull, "w") as devnull:
		proc = subprocess.Popen(['openssl'] + args, stdin=subprocess.PIPE, stderr=devnull)
		for i in range(6):
			proc.stdin.write("\n")
		proc.stdin.close()
		proc.communicate()
def setupSkyped():
	try:
		shutil.rmtree("t/skyped")
	except OSError:
		pass
	os.makedirs("t/skyped")
	cwd = os.getcwd()
	os.chdir("t/skyped")
	try:
		shutil.copyfile("../../skyped.cnf", "skyped.cnf")
		openssl(['req', '-new', '-x509', '-days', '365', '-nodes', '-config', 'skyped.cnf', '-out', 'skyped.cert.pem', '-keyout', 'skyped.key.pem'])
		with open("skyped.conf", "w") as sock:
			sock.write("[skyped]\n")
			sock.write("username = alice\n")
			sock.write("password = %s\n" % hashlib.sha1("foo").hexdigest())
			sock.write("cert = %s/skyped.cert.pem\n" % os.getcwd())
			sock.write("key = %s/skyped.key.pem\n" % os.getcwd())
			sock.write("port = 2727\n")
	finally:
		os.chdir(cwd)

class Test(unittest.TestCase):
	def mock(self, name):
		with open("t/skyped.log", "w") as skyped_log,\
				open("t/pexpect.log", "w") as pexpect_log:
			skyped = subprocess.Popen([sys.executable, "skyped.py",
				"-c", "t/skyped/skyped.conf", "-n", "-d", "-m", "t/%s-skyped.mock" % name],
				stdout=skyped_log, stderr=subprocess.STDOUT)
			try:
				bitlbee = pexpect.spawn('../../bitlbee', ['-d', 't/bitlbee'], logfile=pexpect_log)
				if os.environ.get('ATTACH_GDB'):
					subprocess.Popen(['gdb', '-batch-silent',
						'-ex', 'set logging overwrite on',
						'-ex', 'set logging file t/gdb-%s.log' % bitlbee.pid,
						'-ex', 'set logging on',
						'-ex', 'handle all pass nostop noprint',
						'-ex', 'handle SIGSEGV pass stop print',
						'-ex', 'set pagination 0',
						'-ex', 'continue',
						'-ex', 'backtrace full',
						'-ex', 'info registers',
						'-ex', 'thread apply all backtrace',
						'-ex', 'quit',
						'../../bitlbee', str(bitlbee.pid) ])
				bitlbee_mock = open("t/%s-bitlbee.mock" % name)
				for i in bitlbee_mock.readlines():
					line = i.strip()
					if line.startswith(">> "):
						bitlbee.expect_exact(line[3:], timeout=10)
					elif line.startswith("<< "):
						bitlbee.sendline(line[3:])
				bitlbee_mock.close()
				bitlbee.close()
			finally:
				skyped.terminate()
				skyped.communicate()

	def setUp(self):
		try:
			shutil.rmtree("t/bitlbee")
		except OSError:
			pass
		os.makedirs("t/bitlbee")

	def testMsg(self):
		self.mock("msg")

	def testLogin(self):
		self.mock("login")

	def testInfo(self):
		self.mock("info")

	def testCall(self):
		self.mock("call")

	def testCallFailed(self):
		self.mock("call-failed")

	def testAddYes(self):
		self.mock("add-yes")

	def testAddedYes(self):
		self.mock("added-yes")

	def testAddedNo(self):
		self.mock("added-no")

	def testGroupchatInvited(self):
		self.mock("groupchat-invited")

	def testGroupchatInvite(self):
		self.mock("groupchat-invite")

	def testGroupchatLeave(self):
		self.mock("groupchat-leave")
	
	def testGroupchatMsg(self):
		self.mock("groupchat-msg")
	
	def testGroupchatTopic(self):
		self.mock("groupchat-topic")
	
	def testCalledYes(self):
		self.mock("called-yes")

	def testCalledNo(self):
		self.mock("called-no")

	def testFiletransfer(self):
		self.mock("filetransfer")

	def testGroupRead(self):
		self.mock("group-read")

	def testGroupAdd(self):
		self.mock("group-add")

	def testCtcpHelp(self):
		self.mock("ctcp-help")

	def testSetMoodText(self):
		self.mock("set-mood-text")
	
	def testAwaySet(self):
		self.mock("away-set")

if __name__ == '__main__':
	setupSkyped()
	unittest.main()
">{ char *item; if (!key || !value) { return; } item = g_strdup_printf("%s=%s", key, value); *params = g_slist_insert_sorted(*params, item, (GCompareFunc) strcmp); } void oauth_params_del(GSList **params, const char *key) { int key_len = strlen(key); GSList *l, *n; if (!params) { return; } for (l = *params; l; l = n) { n = l->next; char *data = l->data; if (strncmp(data, key, key_len) == 0 && data[key_len] == '=') { *params = g_slist_remove(*params, data); g_free(data); } } } void oauth_params_set(GSList **params, const char *key, const char *value) { oauth_params_del(params, key); oauth_params_add(params, key, value); } const char *oauth_params_get(GSList **params, const char *key) { int key_len = strlen(key); GSList *l; if (params == NULL) { return NULL; } for (l = *params; l; l = l->next) { if (strncmp((char *) l->data, key, key_len) == 0 && ((char *) l->data)[key_len] == '=') { return (const char *) l->data + key_len + 1; } } return NULL; } void oauth_params_parse(GSList **params, char *in) { char *amp, *eq, *s; while (in && *in) { eq = strchr(in, '='); if (!eq) { break; } *eq = '\0'; if ((amp = strchr(eq + 1, '&'))) { *amp = '\0'; } s = g_strdup(eq + 1); http_decode(s); oauth_params_add(params, in, s); g_free(s); *eq = '='; if (amp == NULL) { break; } *amp = '&'; in = amp + 1; } } void oauth_params_free(GSList **params) { while (params && *params) { g_free((*params)->data); *params = g_slist_remove(*params, (*params)->data); } } char *oauth_params_string(GSList *params) { GSList *l; GString *str = g_string_new(""); for (l = params; l; l = l->next) { char *s, *eq; s = g_malloc(strlen(l->data) * 3 + 1); strcpy(s, l->data); if ((eq = strchr(s, '='))) { http_encode(eq + 1); } g_string_append(str, s); g_free(s); if (l->next) { g_string_append_c(str, '&'); } } return g_string_free(str, FALSE); } void oauth_info_free(struct oauth_info *info) { if (info) { g_free(info->auth_url); g_free(info->request_token); g_free(info->token); g_free(info->token_secret); oauth_params_free(&info->params); g_free(info); } } static void oauth_add_default_params(GSList **params, const struct oauth_service *sp) { char *s; oauth_params_set(params, "oauth_consumer_key", sp->consumer_key); oauth_params_set(params, "oauth_signature_method", "HMAC-SHA1"); s = g_strdup_printf("%d", (int) time(NULL)); oauth_params_set(params, "oauth_timestamp", s); g_free(s); s = oauth_nonce(); oauth_params_set(params, "oauth_nonce", s); g_free(s); oauth_params_set(params, "oauth_version", "1.0"); } static void *oauth_post_request(const char *url, GSList **params_, http_input_function func, struct oauth_info *oi) { GSList *params = NULL; char *s, *params_s, *post; void *req; url_t url_p; if (!url_set(&url_p, url)) { oauth_params_free(params_); return NULL; } if (params_) { params = *params_; } oauth_add_default_params(&params, oi->sp); params_s = oauth_params_string(params); oauth_params_free(&params); s = oauth_sign("POST", url, params_s, oi); post = g_strdup_printf("%s&oauth_signature=%s", params_s, s); g_free(params_s); g_free(s); s = g_strdup_printf("POST %s HTTP/1.0\r\n" "Host: %s\r\n" "Content-Type: application/x-www-form-urlencoded\r\n" "Content-Length: %zd\r\n" "\r\n" "%s", url_p.file, url_p.host, strlen(post), post); g_free(post); req = http_dorequest(url_p.host, url_p.port, url_p.proto == PROTO_HTTPS, s, func, oi); g_free(s); return req; } static void oauth_request_token_done(struct http_request *req); struct oauth_info *oauth_request_token(const struct oauth_service *sp, oauth_cb func, void *data) { struct oauth_info *st = g_new0(struct oauth_info, 1); GSList *params = NULL; st->func = func; st->data = data; st->sp = sp; oauth_params_add(&params, "oauth_callback", "oob"); if (!oauth_post_request(sp->url_request_token, &params, oauth_request_token_done, st)) { oauth_info_free(st); return NULL; } return st; } static void oauth_request_token_done(struct http_request *req) { struct oauth_info *st = req->data; st->http = req; if (req->status_code == 200) { GSList *params = NULL; st->auth_url = g_strdup_printf("%s?%s", st->sp->url_authorize, req->reply_body); oauth_params_parse(&params, req->reply_body); st->request_token = g_strdup(oauth_params_get(&params, "oauth_token")); st->token_secret = g_strdup(oauth_params_get(&params, "oauth_token_secret")); oauth_params_free(&params); } st->stage = OAUTH_REQUEST_TOKEN; st->func(st); } static void oauth_access_token_done(struct http_request *req); gboolean oauth_access_token(const char *pin, struct oauth_info *st) { GSList *params = NULL; oauth_params_add(&params, "oauth_token", st->request_token); oauth_params_add(&params, "oauth_verifier", pin); return oauth_post_request(st->sp->url_access_token, &params, oauth_access_token_done, st) != NULL; } static void oauth_access_token_done(struct http_request *req) { struct oauth_info *st = req->data; st->http = req; if (req->status_code == 200) { oauth_params_parse(&st->params, req->reply_body); st->token = g_strdup(oauth_params_get(&st->params, "oauth_token")); g_free(st->token_secret); st->token_secret = g_strdup(oauth_params_get(&st->params, "oauth_token_secret")); } st->stage = OAUTH_ACCESS_TOKEN; if (st->func(st)) { /* Don't need these anymore, but keep the rest. */ g_free(st->auth_url); st->auth_url = NULL; g_free(st->request_token); st->request_token = NULL; oauth_params_free(&st->params); } } char *oauth_http_header(struct oauth_info *oi, const char *method, const char *url, char *args) { GSList *params = NULL, *l; char *sig = NULL, *params_s, *s; GString *ret = NULL; oauth_params_add(&params, "oauth_token", oi->token); oauth_add_default_params(&params, oi->sp); /* Start building the OAuth header. 'key="value", '... */ ret = g_string_new("OAuth "); for (l = params; l; l = l->next) { char *kv = l->data; char *eq = strchr(kv, '='); char esc[strlen(kv) * 3 + 1]; if (eq == NULL) { break; /* WTF */ } strcpy(esc, eq + 1); http_encode(esc); g_string_append_len(ret, kv, eq - kv + 1); g_string_append_c(ret, '"'); g_string_append(ret, esc); g_string_append(ret, "\", "); } /* Now, before generating the signature, add GET/POST arguments to params since they should be included in the base signature string (but not in the HTTP header). */ if (args) { oauth_params_parse(&params, args); } if ((s = strchr(url, '?'))) { s = g_strdup(s + 1); oauth_params_parse(&params, s); g_free(s); } /* Append the signature and we're done! */ params_s = oauth_params_string(params); sig = oauth_sign(method, url, params_s, oi); g_string_append_printf(ret, "oauth_signature=\"%s\"", sig); g_free(params_s); oauth_params_free(&params); g_free(sig); return ret ? g_string_free(ret, FALSE) : NULL; } char *oauth_to_string(struct oauth_info *oi) { GSList *params = NULL; char *ret; oauth_params_add(&params, "oauth_token", oi->token); oauth_params_add(&params, "oauth_token_secret", oi->token_secret); ret = oauth_params_string(params); oauth_params_free(&params); return ret; } struct oauth_info *oauth_from_string(char *in, const struct oauth_service *sp) { struct oauth_info *oi = g_new0(struct oauth_info, 1); GSList *params = NULL; oauth_params_parse(&params, in); oi->token = g_strdup(oauth_params_get(&params, "oauth_token")); oi->token_secret = g_strdup(oauth_params_get(&params, "oauth_token_secret")); oauth_params_free(&params); oi->sp = sp; return oi; }