网络摄像头rtsp协议登录认证

rtsp协议默认端口554

数据包请求的时候密码非明文，而是通过了加密

具体加密方式如下：

md5(md5(username:realm:password):nonce:md5(public_method:url))

拆分如下：

第一次A md5(username:realm:password) =》用户名:返回的realm值:密码

第二次B md5(public_method:url))=>  DESCRIBE:视频地址url（不同品牌url不一样）

第三次C A:返回nonce的值:B

C再次进行md5，得到C才是真正登录密码

# -*- coding: UTF-8 -*-

import socket
import time
import threading
import hashlib
import base64
import sys

username = "admin"
password = "123456"
serverIp = sys.argv[1]
url = "rtsp://"+serverIp+":554/video1"
m_Vars = {
    "bufLen" : 1024 * 100,
    "defaultServerIp" : serverIp,
    "defaultServerPort" : 554,
    "defaultTestUrl" : url,
    "defaultUserAgent" : "LibVLC/3.0.8 (LIVE555 Streaming Media v2016.11.28)",
    "GUID":"00000000-0000-0000-0000-000000000000"
}

def genmsg_OPTIONS(GUID ,url,seq,userAgent):
    
    
    msgRet = "OPTIONS " + url + " RTSP/1.0
"
    msgRet += "ClientID: RTSP Player
"
    msgRet += "GUID:" + GUID+ "
"
    msgRet += "CSeq: " + str(seq) + "
"
    msgRet += "User-Agent: " + userAgent + "
"
    msgRet += "
"
    return msgRet

def genmsg_DESCRIBE(url,seq,userAgent):
    msgRet = "DESCRIBE " + url + " RTSP/1.0
"
    msgRet += "CSeq: " + str(seq) + "
"
    msgRet += "User-Agent: " + userAgent + "
"
    msgRet += "Accept: application/sdp
"
    msgRet += "
"
    return msgRet
    
#验证请求 用户名+ 经过加密的密码
def genmsg_DESCRIBE_2(url,seq,userAgent,realm_value,nonce_value,response_value):
    msgRet = "DESCRIBE " + url + " RTSP/1.0
"
    msgRet += "CSeq: " + str(seq) + "
"
    msgRet += 'Authorization: Digest username="admin", realm="' + realm_value + '", nonce="' + nonce_value + '", uri="' + url + '", response="' + response_value + '"
'
    msgRet += "User-Agent: " + userAgent + "
"
    msgRet += "Accept: application/sdp
"
    msgRet += "
"
    return msgRet
#md5(md5(username:realm:password):nonce:md5(public_method:url))
def gen_response_value(url,username,password,realm,nonce):
    
    frist_pre_md5_value = hashlib.md5((username + ':' + realm + ':' + password).encode()).hexdigest()
    first_post_md5_value = hashlib.md5(('DESCRIBE:' + url).encode()).hexdigest()
    response_value = hashlib.md5((frist_pre_md5_value + ':' + nonce + ':' + first_post_md5_value).encode()).hexdigest()
    return response_value



def genmsg_SETUP(url,seq,userAgent):
    msgRet = "SETUP " + url + " RTSP/1.0
"
    msgRet += "CSeq: " + str(seq) + "
"
    msgRet += "User-Agent: " + userAgent + "
"
    msgRet += "Transport: RTP/AVP/TCP;mode=play
"   
    msgRet += "
"
    return msgRet

def genmsg_PLAY(url,seq,userAgent,sessionId):
    msgRet = "PLAY " + url + " RTSP/1.0
"
    msgRet += "CSeq: " + str(seq) + "
"
    msgRet += "User-Agent: " + userAgent + "
"
    msgRet += "Session: " + sessionId + "
"
    msgRet += "
"
    return msgRet

def decodeMsg(strContent):
    tostr = strContent.decode("gb2312")
    mapRetInf = {}        
    for m in [elem for elem in tostr.split("
") if len(elem) > 1][2:-1]:
        #print str
        tmp2 = m.split(":")
        mapRetInf[tmp2[0]]=tmp2[1][:-1]       
    return mapRetInf
    


print("--------------开始---------------")
s = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
s.connect((m_Vars["defaultServerIp"],m_Vars["defaultServerPort"]))    
seq  = 2
#向服务器发送OPTIONS请求，得到服务器所提供的方法
str0 =  genmsg_OPTIONS(m_Vars["GUID"],m_Vars["defaultTestUrl"],seq,m_Vars["defaultUserAgent"])
#print (str0)
s.send(str0.encode(encoding='utf_8', errors='strict'))
s.recv(m_Vars["bufLen"])
seq = seq + 1
#向服务器发送DESCRIBE请求，得到SDP
str1 = genmsg_DESCRIBE(m_Vars["defaultTestUrl"],seq,m_Vars["defaultUserAgent"])
s.send(str1.encode(encoding='utf_8', errors='strict'))
msg1 = s.recv(m_Vars["bufLen"])
#print("
===得到realm和nonce值用于加密===")
#print (msg1)
seq = seq + 1
#获取得到realm和nonce值用于加密
Demsg1 = msg1.decode("utf-8")
realm_pos = Demsg1.find('realm')
realm_value_begin_pos = Demsg1.find('"',realm_pos)+1
realm_value_end_pos = Demsg1.find('"',realm_pos+8)
realm_value = Demsg1[realm_value_begin_pos:realm_value_end_pos]
nonce_pos = Demsg1.find('nonce')
nonce_value_begin_pos = Demsg1.find('"',nonce_pos)+1
nonce_value_end_pos = Demsg1.find('"',nonce_pos+8)
nonce_value = Demsg1[nonce_value_begin_pos:nonce_value_end_pos]
#gen_response_value(url,username,password,realm,nonce)

response_value = gen_response_value(m_Vars["defaultTestUrl"], username, password,realm_value, nonce_value)

#url,seq,userAgent,realm_value,nonce_value,response_value
str2 = genmsg_DESCRIBE_2(m_Vars["defaultTestUrl"],seq,m_Vars["defaultUserAgent"],realm_value,nonce_value,response_value)
#print(str2)
s.send(str2.encode(encoding='utf_8', errors='strict'))
msg2 = s.recv(m_Vars["bufLen"])

print (msg2)

s.close()