عندما جئت للعمل في هذه الشركة، كان لدي بالفعل بعض قواعد البيانات الخاصة بأجهزة IP، والعديد من الخوادم ذات العلامات النجمية والتصحيح على شكل FreeBPX. بالإضافة إلى ذلك، يعمل PBX Samsung IDCS500 التناظري بالتوازي، وبشكل عام، كان نظام الاتصال الرئيسي في الشركة، وعملت الاتصالات الهاتفية عبر بروتوكول الإنترنت فقط لقسم المبيعات. وكان كل شيء سيستمر في الطهي بهذه الطريقة، ولكن في يوم من الأيام صدر مرسوم بنقل الجميع إلى هاتف IP الهاتفي، وتم الاتفاق على المواعيد النهائية، وتم شراء المعدات وبدأ تنفيذ خطة نقل المؤسسة إلى القرن الحادي والعشرين.
أول ما يثير القلق في مثل هذه الحالة هو العدد المتزايد بسرعة من أجهزة الهاتف التي تحتاج إلى إدارة بطريقة أو بأخرى، والشيء الثاني الذي كان مقلقًا للغاية هو دليل الهاتف. إذا كان بإمكان Endpoint Manager مساعدتنا في الإصدار الأول (والذي، بالمناسبة، تم حذفه من أحدث إصدارات FreePBX)، فقد ظهرت بعض الأسئلة مع الكتاب:
- أولاً، كيف يمكن التأكد من دقتها عندما يتغير موقع/سيولة المستخدمين باستمرار؟
- ثانيا، كيفية نزع الطابع الشخصي تماما عن الهواتف. وعدم ملء اسم جهة الاتصال في كل مرة؟
كانت المشكلة مثيرة للاهتمام، ولم يستغرق الحل وقتًا طويلاً للوصول. الآن سأقدم القائمة الكاملة، وبعد ذلك سننظر فيها بالترتيب.
from scapy.all import sniff
from scapy.layers.inet import IP
import mysql.connector
import ldap
import getpass
import tftpy
import requests
import os
import time
from string import replace
def conn_ldap(login):
ad = ldap.initialize('ldap://***.local')
ad.simple_bind_s('voip@***.local', 'password')
basedn = 'OU=IT,DC=***,DC=LOCAL'
basedn_user = 'OU=***,OU=***,DC=***,DC=LOCAL'
scope = ldap.SCOPE_SUBTREE
filterexp = "(&(sAMAccountName=" + login + ")(ObjectClass=person))"
filterexp2 = "(&(ObjectClass=organizationUnit))"
attrlist = ['cn']
attrlist2 = ['OU']
search = ad.search_s(basedn, scope, filterexp, attrlist)
adname = search[0][1]['cn'][0].decode('utf-8')
if adname == ' ':
search = ad.search_s(basedn_user, scope, filterexp2, attrlist2)
for i in range(1, len(search)+1):
group = search[i][1]['ou'][0]
basedn_user2 = 'OU='+group+','+basedn_user
search = ad.search_s(basedn_user2, scope, filterexp, attrlist)
adname = search[0][1]['cn'][0].decode('utf-8')
if adname != ' ':
return adname
adname = search[0][1]['cn'][0].decode('utf-8')
ad.unbind_s()
return adname
def tftp_file_change(config,place,adname,current_account,current_account_password):
client = tftpy.TftpClient("192.168.0.3", 69)
client.download('template.cfg', place)
fileread = open(place, 'r')
line = fileread.readlines()
fileread.close()
line[5] = (('account.1.label = ').encode('utf-8') + adname.encode('utf-8') + 'n')
line[2] = (('account.1.auth_name = ').encode('utf-8') + current_account.encode('utf-8') + 'n')
line[3] = (('account.1.display_name = ').encode('utf-8') + current_account.encode('utf-8') + 'n')
line[6] = (('account.1.password = ').encode('utf-8') + current_account_password[0][0] + 'n')
filewrite = open(place, 'w')
for i in line:
filewrite.write(i)
filewrite.close()
print place
print config
client.upload(config,place)
def get_phone_inform(ipaddr):
fileconf = requests.get('http://admin:admin@'+ipaddr+'/servlet?phonecfg=get[&accounts=1]')
conf = fileconf.text.split('|')
current_account = conf[2]
return current_account
def sniff_frame():
pcapf = sniff(count=1, timeout=70, filter="dst host 192.168.0.3 and port 5060")
if len(pcapf) == 0:
exit()
frame = pcapf[0]
macaddr = frame.src
print macaddr[:8]
if macaddr[:8] != '80:5e:c0':
exit()
ipaddr = frame[0][IP].src
return macaddr, ipaddr
def conn_mysql(query,fquery,macaddr,qwery2):
connect = mysql.connector.connect(host='192.168.0.3', database='voip', user='voip_wr', password='***')
cursor = connect.cursor()
cursor.execute(fquery)
state = cursor.fetchall()
state = bool(state[0][0])
if state == True:
cursor.execute(qwery2)
connect.commit()
connect.close()
else:
cursor.execute(query)
connect.commit()
connect.close()
def check_account(current_account):
connect = mysql.connector.connect(host='192.168.0.3', database='asterisk', user='voip_wr', password='***')
cursor = connect.cursor()
qwery = 'select data from sip where id=' + current_account + ' and keyword="secret";'
cursor.execute(qwery)
password = cursor.fetchall()
if password == ' ':
exit()
else:
return password
if __name__ == '__main__':
macaddr, ipaddr = sniff_frame()
current_account = get_phone_inform(ipaddr)
current_account_password = check_account(current_account)
macaddr = macaddr.replace(':', '')
ipaddr = ipaddr.decode('utf-8')
adname = conn_ldap(getpass.getuser())
query = 'INSERT INTO station (mac, ip, name, number) VALUES (' + '"' + macaddr + '",' + '"' + ipaddr + '",' + '"' + adname + '",' + '"' + get_phone_inform(ipaddr) + '"' + ')'
qwery2 = 'UPDATE station SET ip=' + '"' + ipaddr + '"' + ', name=' + '"' + adname + '"' + ', number=' + '"' + get_phone_inform(ipaddr) + '"' + ' WHERE mac=' + '"' + macaddr + '"'
fquery = 'SELECT EXISTS(SELECT mac FROM voip.station WHERE mac=' + '"' + macaddr + '")'
query = query.encode('utf-8')
fquery = fquery.encode('utf-8')
config = macaddr + '.cfg'
place = os.path.expanduser("~") + "" + "AppDataLocal" + config
conn_mysql(query,fquery,macaddr,qwery2)
tftp_file_change(config,place,adname,current_account,current_account_password)
requests.get('http://admin:admin@'+ipaddr+'/cgi-bin/ConfigManApp.com?key=AutoP')
requests.get('http://admin:admin@'+ipaddr+'/cgi-bin/ConfigManApp.com?key=Reboot')يعمل البرنامج على جهاز الكمبيوتر الخاص بالمستخدم ويعمل بشرط أن يكون الكمبيوتر متصلاً بالشبكة عبر الهاتف، حيث لا يمكن لـ Yealink T19 العمل كبوابة.
أولا، علينا أن نفهم ما إذا كان متصلا؟ وما هو نظام التشغيل Mac وIP الموجود في هاتفنا.
def sniff_frame():
pcapf = sniff(count=1, timeout=70, filter="dst host 192.168.0.3 and port 5060")
if len(pcapf) == 0:
exit()
frame = pcapf[0]
macaddr = frame.src
print macaddr[:8]
if macaddr[:8] != '80:5e:c0':
exit()
ipaddr = frame[0][IP].src
return macaddr, ipaddr
نستخدم هنا وظيفة sniff من إطار عمل scapy، وبمساعدتها نتلقى حزمة UDP محددة مسبقًا، وننتظر 70 ثانية، وإذا لم نلتقط أي شيء، نخرج.
count=1, timeout=70, filter="dst host 192.168.0.3 and port 5060"بعد ذلك، نتأكد من أن الجهاز هو بالفعل Yealink ونعيد القيم الضرورية (ip و mac).
باستخدام طلب خاص، نكتشف الحساب الجاري على الهاتف. للقيام بذلك، يتم تنزيل التكوين الحالي من الهاتف وتحليله.
def get_phone_inform(ipaddr):
fileconf = requests.get('http://admin:admin@'+ipaddr+'/servlet?phonecfg=get[&accounts=1]')
conf = fileconf.text.split('|')
current_account = conf[2]
return current_accountمعرفة كلمة المرور لهذا الحساب. للقيام بذلك، ننتقل إلى الجدول asterisk.sip وحقل البيانات فيه.
def check_account(current_account):
connect = mysql.connector.connect(host='192.168.0.3', database='asterisk', user='voip_wr', password='***')
cursor = connect.cursor()
qwery = 'select data from sip where id=' + current_account + ' and keyword="secret";'
cursor.execute(qwery)
password = cursor.fetchall()
if password == ' ':
exit()
else:
return passwordحسنًا، في المرحلة النهائية، نتصل بـ ldap AD ونستخدم sAMAccountName الذي تم استلامه من خلال الوظيفة getpass.getuser() خذ cn للمستخدم الحالي (والذي يحتوي عادةً على الاسم الكامل للمستخدم).
def conn_ldap(login):
ad = ldap.initialize('ldap://***.local')
ad.simple_bind_s('voip@***.local', 'password')
basedn = 'OU=***,DC=***,DC=LOCAL'
basedn_user = 'OU=***,OU=***,DC=***,DC=LOCAL'
scope = ldap.SCOPE_SUBTREE
filterexp = "(&(sAMAccountName=" + login + ")(ObjectClass=person))"
filterexp2 = "(&(ObjectClass=organizationUnit))"
attrlist = ['cn']
attrlist2 = ['OU']
search = ad.search_s(basedn, scope, filterexp, attrlist)
adname = search[0][1]['cn'][0].decode('utf-8')
if adname == ' ':
search = ad.search_s(basedn_user, scope, filterexp2, attrlist2)
for i in range(1, len(search)+1):
group = search[i][1]['ou'][0]
basedn_user2 = 'OU='+group+','+basedn_user
search = ad.search_s(basedn_user2, scope, filterexp, attrlist)
adname = search[0][1]['cn'][0].decode('utf-8')
if adname != ' ':
return adname
adname = search[0][1]['cn'][0].decode('utf-8')
ad.unbind_s()
return adnameنتصل بجدول تم إنشاؤه مسبقًا في قاعدة البيانات (لقد قمت بإنشائه هناك) وندخل كل ما تعلمناه، أي: ip، mac، اسم المستخدم.
def conn_mysql(query,fquery,macaddr,qwery2):
connect = mysql.connector.connect(host='192.168.0.3', database='voip', user='voip_wr', password='***')
cursor = connect.cursor()
cursor.execute(fquery)
state = cursor.fetchall()
state = bool(state[0][0])
if state == True:
cursor.execute(qwery2)
connect.commit()
connect.close()
else:
cursor.execute(query)
connect.commit()
connect.close()يمكننا أن نتوقف هنا، لأننا أنشأنا بالفعل دفتر عناوين ديناميكيًا، قد تسأل، لكنني ذهبت إلى أبعد من ذلك وأضفت التوفير التلقائي للأجهزة هنا.
للقيام بذلك، يتم تنزيل تكوين القالب من خادم tftp الذي تم تكوينه مسبقًا، حيث نقوم بإجراء تغييراتنا وحفظه بتنسيق mac.cfg. أي أنه بالنسبة لـ Yealink هناك نوعان من التكوين، أحدهما عالمي، والثاني ينطبق على هاتف معين ويجب أن يكون على شكل mac_phone.cfg
بعد كل التغييرات في الملف وحفظه مرة أخرى على خادم tftp، نعطي الأمر للهاتف لتوفير الجهاز وإعادة تشغيله.
def tftp_file_change(config,place,adname,current_account,current_account_password):
client = tftpy.TftpClient("192.168.0.3", 69)
client.download('template.cfg', place)
fileread = open(place, 'r')
line = fileread.readlines()
fileread.close()
line[5] = (('account.1.label = ').encode('utf-8') + adname.encode('utf-8') + 'n')
line[2] = (('account.1.auth_name = ').encode('utf-8') + current_account.encode('utf-8') + 'n')
line[3] = (('account.1.display_name = ').encode('utf-8') + current_account.encode('utf-8') + 'n')
line[6] = (('account.1.password = ').encode('utf-8') + current_account_password[0][0] + 'n')
filewrite = open(place, 'w')
for i in line:
filewrite.write(i)
filewrite.close()
print place
print config
client.upload(config,place)requests.get('http://admin:admin@'+ipaddr+'/cgi-bin/ConfigManApp.com?key=AutoP')
requests.get('http://admin:admin@'+ipaddr+'/cgi-bin/ConfigManApp.com?key=Reboot')بعد إعادة تشغيل الجهاز، نحصل على اسمنا الكامل على شاشة الهاتف + دفتر عناوين مملوء بشكل صحيح دائمًا في شكل قاعدة بيانات، ثم كل ما تبقى هو إضافة XML وقليل من PHP لعرض المحتوى ديناميكيًا. هناك الكثير من هذه الأمثلة، حتى أن YEALINK نفسها لديها هذه الأمثلة.
ملاحظة: لمزيد من قابلية التوسع، يمكنك نقل الإعدادات الرئيسية (المتغيرات) إلى ملف منفصل.
المصدر: www.habr.com