草庐IT

基与python的GRPC SSL/TLS加密及Token鉴权

学海无涯苦作舟 2023-03-28 原文

目录结构

.
├── protos
│   └── hello_world.proto
├── dists
│   ├── __init__.py
│   ├── hello_world_pb2.py
│   └── hello_world_pb2_grpc.py
├── server.py
├── client.py
├── token.py
├── grpc_ssl_key.pem
├── grpc_ssl_cert.pem
└── grpc_ssl_server.pem

 构建protobuf

# protos/hello_world.proto

syntax = "proto3";

package test;

message User {
    string name = 1;
}

message Result {
    string reply = 1;
}

service Test {
    rpc Hello(User) returns (Result) {}
}

# 通过命令构建
python -m grpc_tools.protoc \
  -I protos \
  --python_out=dists \
  --grpc_python_out=dists \
  protos/hello_world.proto

 SSL/TLS加密

创建密钥

1 // 创建grpc_ssl_key.pem和grpc_ssl_cert.pem
2 // 其中<domain>务必事先指定, 后续需要用到
3 openssl req -subj "/CN=black-ip.yazx.com" -x509 -newkey rsa:4096 -days 7200 \
4     -keyout grpc-ssl-key.pem \
5     -out grpc-ssl-cert.pem
6 // 创建grpc_ssl_server.pem
7 openssl rsa -in grpc-ssl-key.pem -out grpc-ssl-server.pem

 服务端加载密钥

# server.py

from grpc.experimental import aio
from typing import AnyStr
from dist import hello_world_pb2
from dist import hello_world_pb2_grpc

# 实现具体的grpc函数
class TestServicer(hello_world_pb2_grpc.TestServicer):

    async def Hello(self, request, context) -> hello_world_pb2.Result:
        return hello_world_pb2.Result(reply=f"hello {request.name}")

# 读取密钥二进制数据
def read_file(path: str, binary: bool) -> AnyStr:
    with open(path, 'rb' if binary else 'r') as f:
        return f.read()

# 创建grpc服务端证书
def create_ssl_server_credentials() -> aio.grpc.ChannelCredentials:
    return aio.grpc.ssl_server_credentials(
        private_key_certificate_chain_pairs=(
            (
                read_file('grpc_ssl_server.pem', True),
                read_file('grpc_ssl_cert.pem', True),
            ),
        )
    )

# 运行grpc服务
async def run(host: str, port: int) -> None:
    server = aio.server()
    server_credetials = create_ssl_server_credentials()
    server.add_secure_port(f'[::]:{port}', server_credetials)
    hello_world_pb2_grpc.add_TestServicer_to_server(TestServicer(), server)
    await server.start()
    await server.wait_for_termination()

 客户端加载密钥

# client.py

from grpc.experimental import aio
from typing import AnyStr
from dist import hello_world_pb2
from dist import hello_world_pb2_grpc

# 创建grpc客户端证书
def create_ssl_channel_credentials() -> aio.grpc.ChannelCredentials:
    return aio.grpc.ssl_channel_credentials(
        root_certificates=read_file('grpc_ssl_cert.pem')
    )

# 运行grpc客户端
async def run(host: str, port: int) -> None:
    options = (('grpc.ssl_target_name_override', '<domain>'),)
    kwargs = {
        'target': f"{host}:{port}",
        'options': (
            ("grpc.lb_policy_name", "round_robin"),  # 自动根据dns域名解析服务列表
            *options
        )
    }
    channel_credentials = create_ssl_channel_credentials()
    creds = aio.grpc.composite_channel_credentials(channel_credentials)
    channel = aio.secure_channel(**kwargs, credentials=creds)
    await channel.channel_ready()
    stub = hello_world_pb2_grpc.TestStub(channel)
    result = await stub.Hello(hello_world_pb2.User(name='world'))

 Token鉴权

grpc强制Token鉴权必须使用SSL/TLS加密

 实现Token校验

# token.py

from typing import Callable, List, Any
from grpc.experimental import aio

class BearerToken(object):
    code: aio.grpc.StatusCode = aio.grpc.StatusCode.UNAUTHENTICATED
    details: str = 'bad bearer token'

    def __init__(self, token: str) -> None:
        self.token = token

    def __call__(self, func) -> Callable:
        async def wrapper(inner_self, request, context: aio.ServicerContext) -> Any:
            metadata = context.invocation_metadata()
            for item in metadata:
                if item[0] == 'authorization' and item[1] == f'Bearer {self.token}':
                    return await func(inner_self, request, context)
            await context.abort(
                code=self.code,
                details=self.details,
                trailing_metadata=metadata
            )

        return wrapper

 服务端校验Token

# server.py

from grpc.experimental import aio
from typing import AnyStr
from dist import hello_world_pb2
from dist import hello_world_pb2_grpc
from token import BearerToken

# 实现具体的grpc函数
class TestServicer(hello_world_pb2_grpc.TestServicer):

    @BearerToken(token='xxx')
    async def Hello(self, request, context) -> hello_world_pb2.Result:
        return hello_world_pb2.Result(reply=f"hello {request.name}")

 客户端注入Token

# client.py

from grpc.experimental import aio

token = 'xxx'

# 创建Token认证
def create_access_token_credentials(cls) -> aio.grpc.CallCredentials:
    return aio.grpc.access_token_call_credentials(token)

# 运行grpc客户端
async def run(host: str, port: int) -> None:
    options = (('grpc.ssl_target_name_override', '<domain>'),)
    kwargs = {
        'target': f"{host}:{port}",
        'options': (
            ("grpc.lb_policy_name", "round_robin"),  # 自动根据dns域名解析服务列表
            *options
        )
    }
    channel_credentials = create_ssl_channel_credentials()
    token_credentials = self.create_access_token_credentials()
    creds = aio.grpc.composite_channel_credentials(
        channel_credentials,
        token_credentials
    )
    channel = aio.secure_channel(**kwargs, credentials=creds)
    await channel.channel_ready()
    stub = hello_world_pb2_grpc.TestStub(channel)
    result = await stub.Hello(hello_world_pb2.User(name='world'))

 

 

 

pythonTokenspancolorstylePython

有关基与python的GRPC SSL/TLS加密及Token鉴权的更多相关文章

  1. python - 如何使用 Ruby 或 Python 创建一系列高音调和低音调的蜂鸣声？ - 2

    关闭。这个问题是opinion-based.它目前不接受答案。想要改进这个问题？更新问题，以便editingthispost可以用事实和引用来回答它.关闭4年前。Improvethisquestion我想在固定时间创建一系列低音和高音调的哔哔声。例如:在150毫秒时发出高音调的蜂鸣声在151毫秒时发出低音调的蜂鸣声200毫秒时发出低音调的蜂鸣声250毫秒的高音调蜂鸣声有没有办法在Ruby或Python中做到这一点？我真的不在乎输出编码是什么(.wav、.mp3、.ogg等等)，但我确实想创建一个输出文件。

  2. ruby-on-rails - Rails 中的 NoMethodError::MailersController#preview undefined method `activation_token=' for nil:NilClass - 2

    似乎无法为此找到有效的答案。我正在阅读Rails教程的第10章第10.1.2节，但似乎无法使邮件程序预览正常工作。我发现处理错误的所有答案都与教程的不同部分相关，我假设我犯的错误正盯着我的脸。我已经完成并将教程中的代码复制/粘贴到相关文件中，但到目前为止，我还看不出我输入的内容与教程中的内容有什么区别。到目前为止，建议是在函数定义中添加或删除参数user，但这并没有解决问题。触发错误的url是http://localhost:3000/rails/mailers/user_mailer/account_activation.http://localhost:3000/rails/mai

  3. jquery - 我的 jquery AJAX POST 请求无需发送 Authenticity Token (Rails) - 2

    rails中是否有任何规定允许站点的所有AJAXPOST请求在没有authenticity_token的情况下通过？我有一个调用Controller方法的JqueryPOSTajax调用，但我没有在其中放置任何真实性代码，但调用成功。我的ApplicationController确实有'request_forgery_protection'并且我已经改变了config.action_controller.consider_all_requests_local在我的environments/development.rb中为false我还搜索了我的代码以确保我没有重载ajaxSend来发送

  4. Python 相当于 Perl/Ruby ||= - 2

    这个问题在这里已经有了答案:关闭10年前。PossibleDuplicate:Pythonconditionalassignmentoperator对于这样一个简单的问题表示歉意，但是谷歌搜索||=并不是很有帮助；)Python中是否有与Ruby和Perl中的||=语句等效的语句？例如:foo="hey"foo||="what"#assignfooifit'sundefined#fooisstill"hey"bar||="yeah"#baris"yeah"另外，类似这样的东西的通用术语是什么？条件分配是我的第一个猜测，但Wikipediapage跟我想的不太一样。

  5. java - 什么相当于 ruby​​ 的 rack 或 python 的 Java wsgi？ - 2

    什么是ruby​​的rack或python的Java的wsgi？还有一个路由库。 最佳答案 来自Python标准PEP333:Bycontrast,althoughJavahasjustasmanywebapplicationframeworksavailable,Java's"servlet"APImakesitpossibleforapplicationswrittenwithanyJavawebapplicationframeworktoruninanywebserverthatsupportstheservletAPI.ht

  6. 华为OD机试用Python实现 -【明明的随机数】 2023Q1A - 2

    华为OD机试题本篇题目：明明的随机数题目输入描述输出描述：示例1输入输出说明代码编写思路最近更新的博客华为od2023|什么是华为od，od薪资待遇，od机试题清单华为OD机试真题大全，用Python解华为机试题|机试宝典【华为OD机试】全流程解析+经验分享,题型分享,防作弊指南华为o

  7. python - 如何读取 MIDI 文件、更改其乐器并将其写回？ - 2

    我想解析一个已经存在的.mid文件，改变它的乐器，例如从“acousticgrandpiano”到“violin”，然后将它保存回去或作为另一个.mid文件。根据我在文档中看到的内容，该乐器通过program_change或patch_change指令进行了更改，但我找不到任何在已经存在的MIDI文件中执行此操作的库.他们似乎都只支持从头开始创建的MIDI文件。 最佳答案 MIDIpackage会为您完成此操作，但具体方法取决于midi文件的原始内容。一个MIDI文件由一个或多个音轨组成，每个音轨是十六个channel中任何一个上的

  8. 「Python｜Selenium｜场景案例」如何定位iframe中的元素？ - 2

    本文主要介绍在使用Selenium进行自动化测试或者任务时，对于使用了iframe的页面，如何定位iframe中的元素文章目录场景描述解决方案具体代码场景描述当我们在使用Selenium进行自动化测试的时候，可能会遇到一些界面或者窗体是使用HTML的iframe标签进行承载的。对于iframe中的标签，如果直接查找是无法找到的，会抛出没有找到元素的异常。比如近在咫尺的例子就是，CSDN的登录窗体就是使用的iframe，大家可以尝试通过F12开发者模式查看到的tag_name,class_name,id或者xpath来定位中的页面元素，会抛出NoSuchElementException异常。解决

  9. python ffmpeg 使用 pyav 转换 一组图像 到 视频 - 2

    2022/8/4更新支持加入水印水印必须包含透明图像，并且水印图像大小要等于原图像的大小pythonconvert_image_to_video.py-f30-mwatermark.pngim_dirout.mkv2022/6/21更新让命令行参数更加易用新的命令行使用方法pythonconvert_image_to_video.py-f30im_dirout.mkvFFMPEG命令行转换一组JPG图像到视频时，是将这组图像视为MJPG流。我需要转换一组PNG图像到视频，FFMPEG就不认了。pyav内置了ffmpeg库，不需要系统带有ffmpeg工具因此我使用ffmpeg的python包装p

  10. Python 刷Leetcode题库，顺带学英语单词（31） - 2

    ValidPalindromeGivenastring,determineifitisapalindrome,consideringonlyalphanumericcharactersandignoringcases. [#125]Example:"Aman,aplan,acanal:Panama"isapalindrome."raceacar"isnotapalindrome.Haveyouconsiderthatthestringmightbeempty?Thisisagoodquestiontoaskduringaninterview.Forthepurposeofthisproblem

随机推荐