Как программно управлять WiFi маршрутизатором TP-Link с помощью Python requests

в 11:32, , рубрики: internet of things, python, requests, tp-link wr841n

Однажды передо мной встала задача реализации программного управления одним из распространенных домашних Wi-Fi маршрутизаторов TP-Link TL-WR841N, у которого, к сожалению, нет интерфейса управления через командную строку (telnet, SSH). Я хотел, чтобы мой Telegram бот, реализованный на Python на базе SBC в локальной домашней сети, на основе моих команд выполнял следующие функции управления маршрутизатором:

  • Перезагрузка маршрутизатора
  • Открытие/закрытие NAT Port Forwarding к внутренним WEB-сервисам
  • Открытие/закрытие удаленного доступа к маршрутизатору из WAN (интернет)
  • Определение устройств, зарегистрированных в локальной WiFi сети маршрутизатора


Конечно, все это пользователь может выполнять и сам вручную с помощью WEB-интерфейса, но, во-первых, мне хотелось автоматизировать эти функции, во-вторых, иногда это нужно делать удаленно, а я не хотел постоянно держать открытым доступ к управлению маршрутизатором по незашифрованному HTTP через интернет по соображениям безопасности. Управление с помощью «закрытого» Telegram бота мне показалось более надежным.

Для доступа к управлению маршрутизатором я использовал единственный доступный интерфейс — пользовательский WEB-интерфейс, взаимодействие с которым реализовал с помощью HTTP запросов Python requests. Для того, чтобы определить, какие HTTP GET запросы необходимо направлять на маршрутизатор, я использовал всем известный сниффер трафика Wireshark. Проще говоря, я с помощью Python requests воспроизвел те запросы, которые увидел в Wireshark, в необходимой последовательности.

Импорт, авторизация и исходные параметры

Итак, прежде всего импортируем в Python библиотеку requests, на базе которой мы будем реализовывать HTTP запросы.

import requests

В качестве исходных параметров укажем внутренний IP адрес маршрутизатора в виде строки HTTP запроса. В моем случае это 192.168.0.1.

router_ip='http://192.168.0.1'

Для авторизации нам понадобится токен, вычисление которого осуществляется на основе логина и пароля пользователя. Функция вычисления токена задана в JS скрипте на маршрутизаторе 192.168.0.1/login/encrypt.js. Выглядит он вот так.

encrypt.js

function hex_md5(s)
{ 
	return binl2hex(core_md5(str2binl(s), s.length * 8));
}

function core_md5(x, len)
{
  /* append padding */
  x[len >> 5] |= 0x80 << ((len) % 32);
  x[(((len + 64) >>> 9) << 4) + 14] = len;

  var a =  1732584193;
  var b = -271733879;
  var c = -1732584194;
  var d =  271733878;

  for(var i = 0; i < x.length; i += 16)
  {
    var olda = a;
    var oldb = b;
    var oldc = c;
    var oldd = d;

    a = md5_ff(a, b, c, d, x[i+ 0], 7 , -680876936);
    d = md5_ff(d, a, b, c, x[i+ 1], 12, -389564586);
    c = md5_ff(c, d, a, b, x[i+ 2], 17,  606105819);
    b = md5_ff(b, c, d, a, x[i+ 3], 22, -1044525330);
    a = md5_ff(a, b, c, d, x[i+ 4], 7 , -176418897);
    d = md5_ff(d, a, b, c, x[i+ 5], 12,  1200080426);
    c = md5_ff(c, d, a, b, x[i+ 6], 17, -1473231341);
    b = md5_ff(b, c, d, a, x[i+ 7], 22, -45705983);
    a = md5_ff(a, b, c, d, x[i+ 8], 7 ,  1770035416);
    d = md5_ff(d, a, b, c, x[i+ 9], 12, -1958414417);
    c = md5_ff(c, d, a, b, x[i+10], 17, -42063);
    b = md5_ff(b, c, d, a, x[i+11], 22, -1990404162);
    a = md5_ff(a, b, c, d, x[i+12], 7 ,  1804603682);
    d = md5_ff(d, a, b, c, x[i+13], 12, -40341101);
    c = md5_ff(c, d, a, b, x[i+14], 17, -1502002290);
    b = md5_ff(b, c, d, a, x[i+15], 22,  1236535329);

    a = md5_gg(a, b, c, d, x[i+ 1], 5 , -165796510);
    d = md5_gg(d, a, b, c, x[i+ 6], 9 , -1069501632);
    c = md5_gg(c, d, a, b, x[i+11], 14,  643717713);
    b = md5_gg(b, c, d, a, x[i+ 0], 20, -373897302);
    a = md5_gg(a, b, c, d, x[i+ 5], 5 , -701558691);
    d = md5_gg(d, a, b, c, x[i+10], 9 ,  38016083);
    c = md5_gg(c, d, a, b, x[i+15], 14, -660478335);
    b = md5_gg(b, c, d, a, x[i+ 4], 20, -405537848);
    a = md5_gg(a, b, c, d, x[i+ 9], 5 ,  568446438);
    d = md5_gg(d, a, b, c, x[i+14], 9 , -1019803690);
    c = md5_gg(c, d, a, b, x[i+ 3], 14, -187363961);
    b = md5_gg(b, c, d, a, x[i+ 8], 20,  1163531501);
    a = md5_gg(a, b, c, d, x[i+13], 5 , -1444681467);
    d = md5_gg(d, a, b, c, x[i+ 2], 9 , -51403784);
    c = md5_gg(c, d, a, b, x[i+ 7], 14,  1735328473);
    b = md5_gg(b, c, d, a, x[i+12], 20, -1926607734);

    a = md5_hh(a, b, c, d, x[i+ 5], 4 , -378558);
    d = md5_hh(d, a, b, c, x[i+ 8], 11, -2022574463);
    c = md5_hh(c, d, a, b, x[i+11], 16,  1839030562);
    b = md5_hh(b, c, d, a, x[i+14], 23, -35309556);
    a = md5_hh(a, b, c, d, x[i+ 1], 4 , -1530992060);
    d = md5_hh(d, a, b, c, x[i+ 4], 11,  1272893353);
    c = md5_hh(c, d, a, b, x[i+ 7], 16, -155497632);
    b = md5_hh(b, c, d, a, x[i+10], 23, -1094730640);
    a = md5_hh(a, b, c, d, x[i+13], 4 ,  681279174);
    d = md5_hh(d, a, b, c, x[i+ 0], 11, -358537222);
    c = md5_hh(c, d, a, b, x[i+ 3], 16, -722521979);
    b = md5_hh(b, c, d, a, x[i+ 6], 23,  76029189);
    a = md5_hh(a, b, c, d, x[i+ 9], 4 , -640364487);
    d = md5_hh(d, a, b, c, x[i+12], 11, -421815835);
    c = md5_hh(c, d, a, b, x[i+15], 16,  530742520);
    b = md5_hh(b, c, d, a, x[i+ 2], 23, -995338651);

    a = md5_ii(a, b, c, d, x[i+ 0], 6 , -198630844);
    d = md5_ii(d, a, b, c, x[i+ 7], 10,  1126891415);
    c = md5_ii(c, d, a, b, x[i+14], 15, -1416354905);
    b = md5_ii(b, c, d, a, x[i+ 5], 21, -57434055);
    a = md5_ii(a, b, c, d, x[i+12], 6 ,  1700485571);
    d = md5_ii(d, a, b, c, x[i+ 3], 10, -1894986606);
    c = md5_ii(c, d, a, b, x[i+10], 15, -1051523);
    b = md5_ii(b, c, d, a, x[i+ 1], 21, -2054922799);
    a = md5_ii(a, b, c, d, x[i+ 8], 6 ,  1873313359);
    d = md5_ii(d, a, b, c, x[i+15], 10, -30611744);
    c = md5_ii(c, d, a, b, x[i+ 6], 15, -1560198380);
    b = md5_ii(b, c, d, a, x[i+13], 21,  1309151649);
    a = md5_ii(a, b, c, d, x[i+ 4], 6 , -145523070);
    d = md5_ii(d, a, b, c, x[i+11], 10, -1120210379);
    c = md5_ii(c, d, a, b, x[i+ 2], 15,  718787259);
    b = md5_ii(b, c, d, a, x[i+ 9], 21, -343485551);

    a = safe_add(a, olda);
    b = safe_add(b, oldb);
    c = safe_add(c, oldc);
    d = safe_add(d, oldd);
  }
  return Array(a, b, c, d);

}

function md5_cmn(q, a, b, x, s, t)
{
  return safe_add(bit_rol(safe_add(safe_add(a, q), safe_add(x, t)), s),b);
}
function md5_ff(a, b, c, d, x, s, t)
{
  return md5_cmn((b & c) | ((~b) & d), a, b, x, s, t);
}
function md5_gg(a, b, c, d, x, s, t)
{
  return md5_cmn((b & d) | (c & (~d)), a, b, x, s, t);
}
function md5_hh(a, b, c, d, x, s, t)
{
  return md5_cmn(b ^ c ^ d, a, b, x, s, t);
}
function md5_ii(a, b, c, d, x, s, t)
{
  return md5_cmn(c ^ (b | (~d)), a, b, x, s, t);
}

/*
 * Add integers, wrapping at 2^32. This uses 16-bit operations internally
 * to work around bugs in some JS interpreters.
 */
function safe_add(x, y)
{
  var lsw = (x & 0xFFFF) + (y & 0xFFFF);
  var msw = (x >> 16) + (y >> 16) + (lsw >> 16);
  return (msw << 16) | (lsw & 0xFFFF);
}

function bit_rol(num, cnt)
{
  return (num << cnt) | (num >>> (32 - cnt));
}

function str2binl(str)
{
  var bin = Array();
  var mask = (1 << 8) - 1;
  for(var i = 0; i < str.length * 8; i += 8)
    bin[i>>5] |= (str.charCodeAt(i / 8) & mask) << (i%32);
  return bin;
}

function binl2hex(binarray)
{
  var hex_tab = "0123456789abcdef";
  var str = "";
  for(var i = 0; i < binarray.length * 4; i++)
  {
    str += hex_tab.charAt((binarray[i>>2] >> ((i%4)*8+4)) & 0xF) +
           hex_tab.charAt((binarray[i>>2] >> ((i%4)*8  )) & 0xF);
  }
  return str;
}

function Base64Encoding(input) 
{
	var keyStr = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/=";
	var output = "";
	var chr1, chr2, chr3, enc1, enc2, enc3, enc4;
	var i = 0;

	//input = utf8_encode(input);

	while (i < input.length) 
	{

		chr1 = input.charCodeAt(i++);
		chr2 = input.charCodeAt(i++);
		chr3 = input.charCodeAt(i++);

		enc1 = chr1 >> 2;
		enc2 = ((chr1 & 3) << 4) | (chr2 >> 4);
		enc3 = ((chr2 & 15) << 2) | (chr3 >> 6);
		enc4 = chr3 & 63;

		if (isNaN(chr2)) {
			enc3 = enc4 = 64;
		} else if (isNaN(chr3)) {
			enc4 = 64;
		}

		output = output +
		keyStr.charAt(enc1) + keyStr.charAt(enc2) +
		keyStr.charAt(enc3) + keyStr.charAt(enc4);

	}
 
	return output;
}

function utf8_encode (string) 
{
	string = string.replace(/rn/g,"n");
	var utftext = "";

	for (var n = 0; n < string.length; n++) {

		var c = string.charCodeAt(n);

		if (c < 128) {
			utftext += String.fromCharCode(c);
		}
		else if((c > 127) && (c < 2048)) {
			utftext += String.fromCharCode((c >> 6) | 192);
			utftext += String.fromCharCode((c & 63) | 128);
		}
		else {
			utftext += String.fromCharCode((c >> 12) | 224);
			utftext += String.fromCharCode(((c >> 6) & 63) | 128);
			utftext += String.fromCharCode((c & 63) | 128);
		}

	}

	return utftext;
}

Но, признаюсь, я не стал портировать эту функцию из JS в Python. Я упростил подход и посмотрел значение параметра, которое отправляет мой браузер в Wireshark.

Как программно управлять WiFi маршрутизатором TP-Link с помощью Python requests - 1

Выделенный запрос — это запрос авторизации, который мы воспроизведем чуть позже. Таким образом, я скопировал параметр Cookie pair в запросе авторизации

auth_token='Authorization=Basic%20YWRtaW46YjAxYzZmYzYyMDgwMzA5Y2ZiMzc2ZTE4NzI3YzMwNzk%3D'

Пожалуй, это не самый изящный, но простой подход. Кроме того, мы не будем хранить логин и пароль от маршрутизатора в текстовом открытом виде, что, на мой взгляд, не так плохо.

Перейдем к функции авторизации.

def login():
    r = requests.get(router_ip+'/userRpm/LoginRpm.htm?Save=Save',headers={'Referer':router_ip+'/','Cookie': auth_token})

    if r.status_code==200:
        x=1
        while x<3:
            try:
                session_id=r.text[r.text.index(router_ip)+len(router_ip)+1:r.text.index('userRpm')-1]
                return session_id
                break
            except ValueError:
                return 'Login error'
            x+=1
    else:
        return 'IP unreachable'

Цикл while я использовал, так как маршрутизатор не всегда авторизовывал моего бота с первого раза. Возможно, у вас получится обойтись без него.

В случае успешной авторизации функция возвращает значение session_id. Это идентификатор сессии, который генерирует маршрутизатор при авторизации. После прохождения авторизации session_id должен присутствовать во всех последующих HTTP запросах к маршрутизатору.

Далее реализуем функцию выхода logout.

def logout(session_id):
    r = requests.get(router_ip+'/'+session_id+'/userRpm/LogoutRpm.htm',headers={'Referer':router_ip+'/'+session_id+'/userRpm/MenuRpm.htm','Cookie': auth_token})  
    if r.status_code==200:
        return 'Loging out: '+str(r.status_code)
    else:
        return 'Unable to logout''

Я делаю logout после каждой операции, так как маршрутизатор позволяет одновременно подключаться только одному пользователю. Поэтому, если ваш бот авторизуется и не выйдет из маршрутизатора, то вы не сможете на него зайти до тех пор, пока маршрутизатор не закроет открытую сессию по таймауту через несколько минут. Таким образом, я решил строго придерживаться последовательности «login -> операция -> logout».

Кстати, стоит учесть, что, если кто-то уже авторизован на маршрутизаторе, то бот, очевидно, не сможет авторизоваться до тех пор, пока пользователь не сделает logout, или активная сессия не закроется по таймауту. Одним словом, «кто первый встал, того и тапки.» Стоит отметить, что Python бот выполняет операции управления маршрутизатором за доли секунд. Таким образом, ваш маршрутизатор не будет занят ботом в течении продолжительного времени.

Перезагрузка, NAP Port Forwarding

Перейдем к операциям, которые мы можем выполнять после успешной авторизации.

#Перезагрузка
r = requests.get(router_ip+'/'+session+'/userRpm/SysRebootRpm.htm?Reboot=%D0%9F%D0%B5%D1%80%D0%B5%D0%B7%D0%B0%D0%B3%D1%80%D1%83%D0%B7%D0%B8%D1%82%D1%8C',headers={'Referer':router_ip+'/'+session+'/userRpm/SysRebootRpm.htm','Cookie': auth_token})

#Открыть Port Forwarding
r = requests.get(router_ip+'/'+session+'/userRpm/VirtualServerRpm.htm?doAll=EnAll&Page=1',headers={'Referer':router_ip+'/'+session+'/userRpm/VirtualServerRpm.htm','Cookie': auth_token})

#Закрыть Port Forwarding
r = requests.get(router_ip+'/'+session+'/userRpm/VirtualServerRpm.htm?doAll=DisAll&Page=1',headers={'Referer':router_ip+'/'+session+'/userRpm/VirtualServerRpm.htm','Cookie': auth_token})

Здесь я бы хотел подчеркнуть, что указанные запросы активируют/деактивируют все правила Port Forwarding, созданные ранее в соответствующем разделе управления маршрутизатором.

Как программно управлять WiFi маршрутизатором TP-Link с помощью Python requests - 2

Проще говоря, запросы аналогичны нажатию кнопок «Включить все» и «Отключить все». По аналогии можно реализовать и создание/активацию отдельных правил.

Удаленный доступ к маршрутизатору из WAN (интернет)

#Задание IP адреса удаленного управления
r = requests.get(router_ip+'/'+session+'/userRpm/ManageControlRpm.htm?port=5110&ip='+remote_ip+'&Save=%D0%A1%D0%BE%D1%85%D1%80%D0%B0%D0%BD%D0%B8%D1%82%D1%8C',headers={'Referer':router_ip+'/'+session+'/userRpm/SysRebootRpm.htm','Cookie': auth_token})

Здесь параметр remote_ip задает IP-адрес удаленного управления, т.е. тот IP адрес, с которого разрешено удаленно заходить на маршрутизатор через интернет.

Как программно управлять WiFi маршрутизатором TP-Link с помощью Python requests - 3

remote_ip=’255.255.255.255’ #открыто для всех.
remote_ip=’0.0.0.0’ #закрыто для всех.

При желании можно указывать конкретный IP адрес, с которого вы хотите удаленно подключиться к маршрутизатору.

Определение списка подключенных устройств

#Определение подключенных устройств

r = requests.get(router_ip+'/'+session+'/userRpm/WlanStationRpm.htm',headers={'Referer':router_ip+'/'+session+'/userRpm/MenuRpm.htm','Cookie': auth_token})

presence='Дома находятся:'
if 'DC-31-54-97-51-06' in r.text:
            presence=presence+'n'+'DC-31-54-97-51-06'

Для чего нужен функционал определения устройств, зарегистрированных в домашней сети WiFi. Допустим, у меня есть смартфон с WiFi и MAC адресом DC-31-54-97-51-06. Когда я прихожу домой, мой смартфон регистрируется в домашней сети WiFi. Таким нехитрым способом я могу отслеживать свое присутствие (а точнее, присутствие своего смартфона) в домашней сети (т.е. дома). Этот функционал позволяет автоматизировать ряд функций, связанных с определением моего присутствия. Например, когда я прихожу домой, отключается детектор движения камеры наблюдения и т.д. Ранее я использовал определение присутствия устройств в сети с помощью Ping'а их IP адресов, но в итоге разочаровался в данном методе, так как смартфоны, как оказалось, неохотно и не всегда отвечают на Ping. Дополнительно я использую сниффер ARP пакетов, реализованный на Python, который позволяет отследить момент регистрации устройства с определенным MAC адресом в сети WiFi.

Итак, r.text в коде выше возвращает среди прочего список MAC адресов устройств, зарегистрированных в WiFi сети маршрутизатора. Что вы будете делать с этим списком, зависит только от вашей фантазии.

var hostList = new Array(
"94-36-44-8F-2F-ED", 5, 23487, 10618, 2, 
"60-46-37-C0-43-FC", 5, 27088, 10126, 2, 
"EF-71-44-63-51-E1", 5, 600, 364, 2, 
"77-25-9D-99-ED-33", 5, 1547, 1722, 2, 
0,0 );

Итак, мне остается только подытожить. Весь код выглядит следующим образом.

router.py

import requests

router_ip='http://192.168.0.1'
auth_token='Authorization=Basic%20YWRtaW46YjAxYzZmYzYyMDgwMzA5Y2ZiMzc2ZTE4NzI3YzMwNzk%3D'

def logout(session_id):
    r = requests.get(router_ip+'/'+session_id+'/userRpm/LogoutRpm.htm',headers={'Referer':router_ip+'/'+session_id+'/userRpm/MenuRpm.htm','Cookie': auth_token})  
    if r.status_code==200:
        return 'Loging out: '+str(r.status_code)
    else:
        return 'Unnable to logout'

def login():

    r = requests.get(router_ip+'/userRpm/LoginRpm.htm?Save=Save',headers={'Referer':router_ip+'/','Cookie': auth_token})

    if r.status_code==200:
        x=1
        while x<3:
            try:
                session_id=r.text[r.text.index(router_ip)+len(router_ip)+1:r.text.index('userRpm')-1]
                return session_id
                break

            except ValueError:
                return 'Login error'
            
            x+=1
    else:
        return 'IP unreachable'

def routercontrol(operation,remote_ip='255.255.255.255'):
 
        #Авторизация

    if login()=='IP unreachable' or login()=='Login error':
        return login()
        exit(0)

    else:
        session=login()
        print ('Login OK: '+session)
 
    if operation=='Enable ports': 

        #Открыть Port Forwarding
        r = requests.get(router_ip+'/'+session+'/userRpm/VirtualServerRpm.htm?doAll=EnAll&Page=1',headers={'Referer':router_ip+'/'+session+'/userRpm/VirtualServerRpm.htm','Cookie': auth_token})
        status=str(r.status_code)
        print (logout(session))
        return 'Enable all ports: '+status+' http://31.207.73.10:8082'
 
    elif operation=='Disable ports':

        #Закрыть Port Forwarding
        r = requests.get(router_ip+'/'+session+'/userRpm/VirtualServerRpm.htm?doAll=DisAll&Page=1',headers={'Referer':router_ip+'/'+session+'/userRpm/VirtualServerRpm.htm','Cookie': auth_token})
        status=str(r.status_code)
        print (logout(session))
        return 'Disable all ports: '+status
 
    elif operation=='Reboot':

        #Перезагрузка
        r = requests.get(router_ip+'/'+session+'/userRpm/SysRebootRpm.htm?Reboot=%D0%9F%D0%B5%D1%80%D0%B5%D0%B7%D0%B0%D0%B3%D1%80%D1%83%D0%B7%D0%B8%D1%82%D1%8C',headers={'Referer':router_ip+'/'+session+'/userRpm/SysRebootRpm.htm','Cookie': auth_token})
        status=str(r.status_code)
        print (logout(session))
        return 'Reboot: '+status
 
    elif operation=='Remote IP':
   
        #Задание IP адреса удаленного управления
        r = requests.get(router_ip+'/'+session+'/userRpm/ManageControlRpm.htm?port=5110&ip='+remote_ip+'&Save=%D0%A1%D0%BE%D1%85%D1%80%D0%B0%D0%BD%D0%B8%D1%82%D1%8C',headers={'Referer':router_ip+'/'+session+'/userRpm/SysRebootRpm.htm','Cookie': auth_token})
        status=str(r.status_code)
        print (logout(session))
        return 'Remote IP '+remote_ip+': '+status

    elif operation=='Check presence':
  
        #Определение подключенных устройств
        r = requests.get(router_ip+'/'+session+'/userRpm/WlanStationRpm.htm',headers={'Referer':router_ip+'/'+session+'/userRpm/MenuRpm.htm','Cookie': auth_token})
        status=str(r.status_code)
        print (logout(session))
        presence='Дома находятся:'

        if 'DC-31-54-97-51-06' in r.text:
            presence=presence+'n'+'DC-31-54-97-51-06'
        return presence 

    else:
        return 'Wrong command'

Очевидно, аналогичным образом можно автоматизировать и другие функции управления устройствами, лишенными командной строки, с помощью доступа к WEB-интерфейсу. Например, я подобным образом с помощью HTTP Post реализовал reboot IP камеры DLink. Спасибо за внимание!

Автор: VikSam

Источник

* - обязательные к заполнению поля


https://ajax.googleapis.com/ajax/libs/jquery/3.4.1/jquery.min.js