利用Python requests库实现cas认证

时间:2021-12-31 10:24:45

1.准备工作-背景知识

1.1 requests库简介:

python有很多可以用来测试接口的模块,个人觉得,requests库是最好用的, 在Robot Framwork里,它的测试库requestsLibrary,也是基于requests写的。

1.1.1 安装:

作为第三方模块,使用前,需要安装,最简单的安装方式如下

pip install requests

1.1.2 语法(简明版):

1)如何发送请求 :

发送get请求:requests.get(url, params,  headers, cookies, verify, allow_redirects...)

发送post请求: requests.post(url, params, data, headers, cookies, verify, allow_redirects...)

*****以上只有url参数是必填的,其他可以根据需要选填

*****除了get和post,还支持其他http请求,比如delete等等,不过在我们的cas认证中,只需要get和post;而且一般的接口测试中,最常用的也就是get和post消息

2)如何得到响应:

假设发送了一个请求:

r=requests.get(url, params,  headers, cookies, verify, allow_redirects...)

那么以下是返回结果:

r.headers  返回的头信息

r.text  返回的主体

r.status_code 返回的状态码

1.1.3 基于requests库,重写get和post请求

为了让requests库更能符合我们的需求,进行了简单的封装,重写了get和post请求发送:

设置了常用的headers参数

sesssion_id作为参数传入

对于post消息,会根据data值,设置Content-Length

#encoding=utf-8

import requests

class RestfulTest(object):
     @staticmethod
     def send_get_request(url, params= None, session_id= None, verify = False , allow_redirects= False):
         #disable warnings
         requests.packages.urllib3.disable_warnings()
         if params == None:
            params = {}
         if session_id == None:
            cookies = {}
         else:
            cookies = {'JSESSIONID':session_id}
         headers = { 
              "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
              "Accept-Language": "en-US,en;q=0.5",
              "Connection": "keep-alive",
              "User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:24.0) Gecko/20100101 Firefox/24.0", 
             }
         get_response = requests.get(url, params=params,  headers=headers, cookies=cookies, verify=verify, allow_redirects=allow_redirects)
         return get_response

     @staticmethod
     def send_post_request(url, params= None, data= None, session_id= None, verify= False, allow_redirects=False):
        #disable warnings
        requests.packages.urllib3.disable_warnings()
        if params == None:
            params = {}
        if data == None:
            data = {}
        if session_id == None:
            cookies = {}
        else:
            cookies = {'JSESSIONID':session_id}
        headers = { 
              "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
              "Accept-Language": "en-US,en;q=0.5",
              "Connection": "keep-alive",
              "User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:24.0) Gecko/20100101 Firefox/24.0",
              "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
              "Content-Length": str(len(data))
             }
        post_response = requests.post(url, params=params, data=data, headers=headers, cookies=cookies, verify=verify, allow_redirects=allow_redirects)
        return post_response

1.2 cas 认证原理

CAS旨在为 Web 应用系统提供一种可靠的单点登录方法

Single Sign On,简称SSO,SSO使得在多个应用系统中,用户只需要登录一次就可以访问所有相互信任的应用系统

从结构上看,CAS 包含两个部分: CAS Server 和 CAS Client。

CAS Server 需要独立部署,主要负责对用户的认证工作;CAS Client 负责处理对客户端受保护资源的访问请求,需要登录时,重定向到 CAS Server

利用Python requests库实现cas认证 
CAS Client 与受保护的客户端应用部署在一起,以 Filter 方式保护受保护的资源。对于访问受保护资源的每个 Web 请求,CAS Client 会分析该请求的 Http 请求中是否包含 Service Ticket,如果没有,则说明当前用户尚未登录,于是将请求重定向到指定好的 CAS Server 登录地址,并传递 Service (也就是要访问的目的资源地址),以便登录成功过后转回该地址

用户在第 3 步中输入认证信息,如果登录成功,CAS Server 随机产生一个相当长度、唯一、不可伪造的 Service Ticket,并缓存以待将来验证,之后系统自动重定向到 Service 所在地址,并为客户端浏览器设置一个 Ticket Granted Cookie(TGC),CAS Client 在拿到 Service 和新产生的 Ticket 过后,在第 5,6 步中与 CAS Server进行身份合适,以确保 Service Ticket 的合法性

在该协议中,所有与 CAS 的交互均采用 SSL 协议,确保ST 和 TGC 的安全性。协议工作过程中会有 2 次重定向的过程,但是 CAS Client 与 CAS Server 之间进行 Ticket 验证的过程对于用户是透明的。

总结一下,如下:

访问服务: SSO 客户端发送请求访问应用系统提供的服务资源。

定向认证: SSO 客户端会重定向用户请求到 SSO 服务器。

用户认证:用户身份认证。

发放票据: SSO 服务器会产生一个随机的 Service Ticket 。

验证票据: SSO 服务器验证票据 Service Ticket 的合法性,验证通过后,允许客户端访问服务。

传输用户信息: SSO 服务器验证票据通过后,传输用户认证结果信息给客户端。

 

2. python代码实现cas认证

以下代码实现的是我们被测server的cas登录,不同产品,获得cookie值方式可能有所不同,post的data值可能也有所不同,但是大致原理是一致的

 

#usr/bin/env python
#encoding=utf-8

from restfulTest import RestfulTest
from dataProcess import DataProcess

class LoginToServer(object):
    
    @staticmethod
    def get_service_sessionId_redirectUrl(service_url):
        response = RestfulTest.send_get_request(service_url)
        cookieValue = response.headers['set-cookie']
        redirectUrl = response.headers['location']
        sessionId = DataProcess.get_first_matching_group_by_pattern(cookieValue, 'JSESSIONID=(\w+); Path')  
        return sessionId, redirectUrl
    
    @staticmethod
    def cas_server_authentication(cas_server_url, userName="admin", userPasswd="admin"):
    
        response1 = RestfulTest.send_get_request(cas_server_url)
        
        cas_server_sessionId = DataProcess.get_first_matching_group_by_pattern(response1.headers['set-cookie'], 'JSESSIONID=(\w+); Path')  
        submitLt = DataProcess.get_first_matching_group_by_pattern(response1.content, 'name="lt" value="(\w+)"')
        http_data_for_auth = "username="+userName+"&password="+userPasswd+"&lt="+submitLt+"&_eventId=submit&submit=Log+In" 
        
        response2 = RestfulTest.send_post_request(cas_server_url, data = http_data_for_auth, session_id = cas_server_sessionId )
  
        acceptLt= DataProcess.get_first_matching_group_by_pattern(response2.content, 'name="lt" value="(\w+)"')
        http_data_for_confirm ="_eventId_accept=Accept&lt="+acceptLt

        response3 = RestfulTest.send_post_request(cas_server_url, data = http_data_for_confirm, session_id = cas_server_sessionId )

        casTicket = DataProcess.get_first_matching_group_by_pattern(response3.headers['set-cookie'], 'CASTGC=(\S+); Path=\/irisCAS; Secure')
        redirectUrl = response3.headers['location']

        return   cas_server_sessionId, casTicket, redirectUrl

    @staticmethod
    def redirect_to_service(service_url, service_sessionId):
        response = RestfulTest.send_get_request(service_url, session_id = service_sessionId, allow_redirects= True)
        return response

    @staticmethod
    def login_to_service(service_url, userName="admin", userPasswd="admin"):
        service_sessionId, redirectToCas = LoginToServer.get_service_sessionId_redirectUrl(service_url)
        cas_server_sessionId, casTicket,  redirectToServer = LoginToServer.cas_server_authentication(redirectToCas, userName, userPasswd)
        response = LoginToServer.redirect_to_service(redirectToServer, service_sessionId)
        if response.status_code == 200:
            print "Login To Service Success"
        else:
            print "Login To Service Fail"
        return service_sessionId