本文整理汇总了Python中secrets.token_bytes方法的典型用法代码示例。如果您正苦于以下问题：Python secrets.token_bytes方法的具体用法？Python secrets.token_bytes怎么用？Python secrets.token_bytes使用的例子？那么恭喜您, 这里精选的方法代码示例或许可以为您提供帮助。您也可以进一步了解该方法所在模块secrets的用法示例。

在下文中一共展示了secrets.token_bytes方法的28个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: generate_secret_with_hash

​点赞 6

​

# 需要导入模块: import secrets [as 别名]

# 或者: from secrets import token_bytes [as 别名]

def generate_secret_with_hash() -> (bytes, bytes):

'''

Generating secret and related ripemd160 hash.

Returns:

tuple: secret, secret hash

Example:

>>> from clove.utils.hashing import generate_secret_with_hash

>>> secret, secret_hash = generate_secret_with_hash()

>>> secret.hex()

'95a968aa18866ffe6ed9ad39a06e27a90e55699734007173bfdd3daa4bab661d'

'''

secret = secrets.token_bytes(32)

secret_hash = hashlib.new('ripemd160', secret).digest()

return secret, secret_hash

开发者ID:Lamden，项目名称:clove，代码行数:18，

示例2: send_browse_request

​点赞 6

​

# 需要导入模块: import secrets [as 别名]

# 或者: from secrets import token_bytes [as 别名]

def send_browse_request(self, search_criteria, key, challenge):

stream = StreamOut(self.settings)

stream.add(search_criteria)

buffer = stream.get()

stream = StreamOut(self.settings)

stream.u8(0) #Packet type

stream.u32(len(buffer))

stream.write(buffer)

if self.settings.get("pia.lan_version") != 0:

self.nonce_counter += 1

stream.u8(self.settings.get("pia.lan_version"))

stream.bool(self.settings.get("pia.crypto_enabled"))

stream.u64(self.nonce_counter)

stream.write(key)

if self.settings.get("pia.crypto_enabled"):

challenge = self.generate_challenge(key, challenge)

else:

challenge = secrets.token_bytes(16 + 256)

stream.write(challenge)

self.s.send(stream.get(), self.broadcast)

开发者ID:Kinnay，项目名称:NintendoClients，代码行数:27，

示例3: encrypt

​点赞 6

​

# 需要导入模块: import secrets [as 别名]

# 或者: from secrets import token_bytes [as 别名]

def encrypt(self, key, settings):

ticket_key = secrets.token_bytes(16)

stream = streams.StreamOut(settings)

stream.datetime(self.expiration)

stream.pid(self.source_pid)

if len(self.session_key) != settings.get("kerberos.key_size"):

raise ValueError("Incorrect session_key length")

stream.write(self.session_key)

ticket_body = stream.get()

final_key = hashlib.md5(key + ticket_key).digest()

kerberos = KerberosEncryption(final_key)

encrypted = kerberos.encrypt(ticket_body)

stream = streams.StreamOut(settings)

stream.buffer(ticket_key)

stream.buffer(encrypted)

return stream.get()

开发者ID:Kinnay，项目名称:NintendoClients，代码行数:23，

示例4: send_packet

​点赞 6

​

# 需要导入模块: import secrets [as 别名]

# 或者: from secrets import token_bytes [as 别名]

def send_packet(self, opcode, payload=b""):

data = bytes([0x80 | opcode])

mask = 0 if self.server_mode else 0x80

length = len(payload)

if length < 126:

data += bytes([mask | length])

elif length <= 0xFFFF:

data += struct.pack(">BH", mask | 0x7E, length)

else:

data += struct.pack(">BQ", mask | 0x7F, length)

if self.server_mode:

data += payload

else:

mask = secrets.token_bytes(4)

data += mask + self.apply_mask(payload, mask)

self.sock.send(data)

开发者ID:Kinnay，项目名称:NintendoClients，代码行数:22，

示例5: generate_ticket

​点赞 6

​

# 需要导入模块: import secrets [as 别名]

# 或者: from secrets import token_bytes [as 别名]

def generate_ticket(self, source, target):

settings = self.settings

user_key = derive_key(source)

server_key = derive_key(target)

session_key = secrets.token_bytes(settings.get("kerberos.key_size"))

internal = kerberos.ServerTicket()

internal.expiration = common.DateTime.fromtimestamp(time.time() + 120)

internal.source_pid = source.pid

internal.session_key = session_key

ticket = kerberos.ClientTicket()

ticket.session_key = session_key

ticket.target_pid = target.pid

ticket.internal = internal.encrypt(server_key, settings)

return ticket.encrypt(user_key, settings)

开发者ID:Kinnay，项目名称:NintendoClients，代码行数:20，

示例6: encrypt_outer

​点赞 6

​

# 需要导入模块: import secrets [as 别名]

# 或者: from secrets import token_bytes [as 别名]

def encrypt_outer(msg, pubkey):

kf = secrets.token_bytes(HASH_SIZE)

com = commit(kf, msg)

send_key = serialization.load_der_public_key(

pubkey,

backend=default_backend(),

)

ctxt = send_key.encrypt(

msg + kf,

padding.OAEP(

mgf=padding.MGF1(algorithm=hashes.SHA256()),

algorithm=hashes.SHA256(),

label=None,

)

)

return (ctxt, com)

开发者ID:fbsamples，项目名称:fbctf-2019-challenges，代码行数:19，

示例7: _initialize_ping_task

​点赞 6

​

# 需要导入模块: import secrets [as 别名]

# 或者: from secrets import token_bytes [as 别名]

def _initialize_ping_task(self):

async def ping():

while not self._pipeline_task.done():

msg = Message("ping", Ping(bytes32(token_bytes(32))))

self.push_message(

OutboundMessage(NodeType.FARMER, msg, Delivery.BROADCAST)

)

self.push_message(

OutboundMessage(NodeType.TIMELORD, msg, Delivery.BROADCAST)

)

self.push_message(

OutboundMessage(NodeType.FULL_NODE, msg, Delivery.BROADCAST)

)

self.push_message(

OutboundMessage(NodeType.HARVESTER, msg, Delivery.BROADCAST)

)

self.push_message(

OutboundMessage(NodeType.WALLET, msg, Delivery.BROADCAST)

)

await asyncio.sleep(self._ping_interval)

return asyncio.create_task(ping())

开发者ID:Chia-Network，项目名称:chia-blockchain，代码行数:24，

示例8: add_private_key_not_extended

​点赞 6

​

# 需要导入模块: import secrets [as 别名]

# 或者: from secrets import token_bytes [as 别名]

def add_private_key_not_extended(self, key_not_extended: PrivateKey):

"""

Creates a new key, and takes only the prefix information (chain code, version, etc).

This is used to migrate pool_sks from keys.yaml, which are not extended. Then adds

the key to the keychain.

"""

key_bytes = bytes(key_not_extended)

new_extended_bytes = bytearray(

bytes(ExtendedPrivateKey.from_seed(token_bytes(32)))

)

final_extended_bytes = bytes(new_extended_bytes[: -len(key_bytes)] + key_bytes)

key = ExtendedPrivateKey.from_bytes(final_extended_bytes)

assert len(final_extended_bytes) == len(new_extended_bytes)

assert key.get_private_key() == key_not_extended

self.add_private_key(key)

开发者ID:Chia-Network，项目名称:chia-blockchain，代码行数:18，

示例9: create_payload

​点赞 6

​

# 需要导入模块: import secrets [as 别名]

# 或者: from secrets import token_bytes [as 别名]

def create_payload(

command: str, data: Dict[str, Any], origin: str, destination: str, string=True

):

response = {

"command": command,

"ack": False,

"data": data,

"request_id": token_bytes().hex(),

"destination": destination,

"origin": origin,

}

if string:

json_str = dict_to_json_str(response)

return json_str

else:

return response

开发者ID:Chia-Network，项目名称:chia-blockchain，代码行数:19，

示例10: reorg_from_index_to_new_index

​点赞 6

​

# 需要导入模块: import secrets [as 别名]

# 或者: from secrets import token_bytes [as 别名]

def reorg_from_index_to_new_index(self, request: ReorgProtocol):

new_index = request.new_index

old_index = request.old_index

coinbase_ph = request.puzzle_hash

top_tip = self.get_tip()

current_blocks = await self.get_current_blocks(top_tip)

block_count = new_index - old_index

more_blocks = bt.get_consecutive_blocks(

self.constants,

block_count,

current_blocks[:old_index],

10,

seed=token_bytes(),

reward_puzzlehash=coinbase_ph,

transaction_data_at_height={},

)

assert self.server is not None

for block in more_blocks:

async for msg in self.respond_block(full_node_protocol.RespondBlock(block)):

self.server.push_message(msg)

self.log.info(f"New message: {msg}")

开发者ID:Chia-Network，项目名称:chia-blockchain，代码行数:25，

示例11: test_request_respond_transaction

​点赞 6

​

# 需要导入模块: import secrets [as 别名]

# 或者: from secrets import token_bytes [as 别名]

def test_request_respond_transaction(self, two_nodes, wallet_blocks):

full_node_1, full_node_2, server_1, server_2 = two_nodes

wallet_a, wallet_receiver, blocks = wallet_blocks

tx_id = token_bytes(32)

request_transaction = fnp.RequestTransaction(tx_id)

msgs = [x async for x in full_node_1.request_transaction(request_transaction)]

assert len(msgs) == 1

assert msgs[0].message.data == fnp.RejectTransactionRequest(tx_id)

receiver_puzzlehash = wallet_receiver.get_new_puzzlehash()

spend_bundle = wallet_a.generate_signed_transaction(

100, receiver_puzzlehash, blocks[2].header.data.coinbase,

)

assert spend_bundle is not None

respond_transaction = fnp.RespondTransaction(spend_bundle)

prop = [x async for x in full_node_1.respond_transaction(respond_transaction)]

assert len(prop) == 1

assert isinstance(prop[0].message.data, fnp.NewTransaction)

request_transaction = fnp.RequestTransaction(spend_bundle.get_hash())

msgs = [x async for x in full_node_1.request_transaction(request_transaction)]

assert len(msgs) == 1

assert msgs[0].message.data == fnp.RespondTransaction(spend_bundle)

开发者ID:Chia-Network，项目名称:chia-blockchain，代码行数:26，

示例12: test_respond_transaction_fail

​点赞 6

​

# 需要导入模块: import secrets [as 别名]

# 或者: from secrets import token_bytes [as 别名]

def test_respond_transaction_fail(self, two_nodes, wallet_blocks):

full_node_1, full_node_2, server_1, server_2 = two_nodes

wallet_a, wallet_receiver, blocks = wallet_blocks

tx_id = token_bytes(32)

request_transaction = fnp.RequestTransaction(tx_id)

msgs = [x async for x in full_node_1.request_transaction(request_transaction)]

assert len(msgs) == 1

assert msgs[0].message.data == fnp.RejectTransactionRequest(tx_id)

receiver_puzzlehash = wallet_receiver.get_new_puzzlehash()

# Invalid transaction does not propagate

spend_bundle = wallet_a.generate_signed_transaction(

100000000000000, receiver_puzzlehash, blocks[3].header.data.coinbase,

)

assert spend_bundle is not None

respond_transaction = fnp.RespondTransaction(spend_bundle)

assert (

len([x async for x in full_node_1.respond_transaction(respond_transaction)])

== 0

)

开发者ID:Chia-Network，项目名称:chia-blockchain，代码行数:24，

示例13: add_group

​点赞 6

​

# 需要导入模块: import secrets [as 别名]

# 或者: from secrets import token_bytes [as 别名]

def add_group(self):

metadata_file = self.args.metadata_file

try:

with open(metadata_file, 'r') as f:

org_metadata = OrganizationMetadata.from_json(json.load(f))

except Exception as e:

print(

"Organization metadata json file not found ,Please check --metadata-file path ")

raise e

payment_storage_client = PaymentStorageClient(self.args.payment_channel_connection_timeout,

self.args.payment_channel_request_timeout, self.args.endpoints)

payment = Payment(self.args.payment_address, self.args.payment_expiration_threshold,

self.args.payment_channel_storage_type, payment_storage_client)

group_id = base64.b64encode(secrets.token_bytes(32))

group = Group(self.args.group_name, group_id.decode("ascii"), payment)

org_metadata.add_group(group)

org_metadata.save_pretty(metadata_file)

开发者ID:singnet，项目名称:snet-cli，代码行数:22，

示例14: load_secret

​点赞 6

​

# 需要导入模块: import secrets [as 别名]

# 或者: from secrets import token_bytes [as 别名]

def load_secret(self, team, namespace) -> bytes:

if team not in self.secrets or namespace not in self.secrets[team]:

secret = secrets.token_bytes(self.SECRET_NUM_BYTES)

try:

# we don't expect self.SECRET_KEY's revision > 0

await self.kvstore.put(

namespace,

self.SECRET_KEY,

bytes_to_str(secret),

revision=1,

team=team,

)

except RevisionError:

res: keybase1.KVGetResult = await self.kvstore.get(

namespace, self.SECRET_KEY, team=team

)

secret = str_to_bytes(res.entry_value)

if team not in self.secrets:

self.secrets[team] = {}

self.secrets[team][namespace] = secret

return self.secrets[team][namespace]

开发者ID:keybase，项目名称:pykeybasebot，代码行数:23，

示例15: test_token_defaults

​点赞 6

​

# 需要导入模块: import secrets [as 别名]

# 或者: from secrets import token_bytes [as 别名]

def test_token_defaults(self):

# Test that token_* functions handle default size correctly.

for func in (secrets.token_bytes, secrets.token_hex,

secrets.token_urlsafe):

with self.subTest(func=func):

name = func.__name__

try:

func()

except TypeError:

self.fail("%s cannot be called with no argument" % name)

try:

func(None)

except TypeError:

self.fail("%s cannot be called with None" % name)

size = secrets.DEFAULT_ENTROPY

self.assertEqual(len(secrets.token_bytes(None)), size)

self.assertEqual(len(secrets.token_hex(None)), 2*size)

开发者ID:bkerler，项目名称:android_universal，代码行数:19，

示例16: test_basic_sharing_random

​点赞 5

​

# 需要导入模块: import secrets [as 别名]

# 或者: from secrets import token_bytes [as 别名]

def test_basic_sharing_random():

secret = secrets.token_bytes(16)

mnemonics = shamir.generate_mnemonics(1, [(3, 5)], secret)[0]

assert shamir.combine_mnemonics(mnemonics[:3]) == shamir.combine_mnemonics(

mnemonics[2:]

)

开发者ID:trezor，项目名称:python-shamir-mnemonic，代码行数:8，

示例17: encrypt

​点赞 5

​

# 需要导入模块: import secrets [as 别名]

# 或者: from secrets import token_bytes [as 别名]

def encrypt(self, iterations: int = 100_000) -> bytes:

"""Encrypt secret."""

salt = secrets.token_bytes(16)

key = self.__derive_key(salt, iterations)

return urlsafe_b64encode(

b"%b%b%b" % (salt,

iterations.to_bytes(4, "big"),

urlsafe_b64decode(Fernet(key).encrypt(self.secret))))

开发者ID:smallwat3r，项目名称:shhh，代码行数:10，

示例18: random_32_byte_hash

​点赞 5

​

# 需要导入模块: import secrets [as 别名]

# 或者: from secrets import token_bytes [as 别名]

def random_32_byte_hash():

"""

Can generate an invoice preimage and corresponding payment hash

:return: 32 byte sha256 hash digest, 32 byte preimage

"""

preimage = token_bytes(32)

_hash = sha256(preimage)

return _hash.digest(), preimage

#########

# Tests #

#########

开发者ID:willcl-ark，项目名称:lnd_grpc，代码行数:15，

示例19: generate_keypair

​点赞 5

​

# 需要导入模块: import secrets [as 别名]

# 或者: from secrets import token_bytes [as 别名]

def generate_keypair() -> Tuple[AccessKey, SecretKey]:

'''

AWS-like access key and secret key generation.

'''

ak = 'AKIA' + base64.b32encode(secrets.token_bytes(10)).decode('ascii')

sk = secrets.token_urlsafe(30)

return AccessKey(ak), SecretKey(sk)

开发者ID:lablup，项目名称:backend.ai-manager，代码行数:9，

示例20: generate_nonce

​点赞 5

​

# 需要导入模块: import secrets [as 别名]

# 或者: from secrets import token_bytes [as 别名]

def generate_nonce() -> bytes:

return secrets.token_bytes(NONCE_LENGTH)

开发者ID:zyv，项目名称:huawei-lpv2，代码行数:4，

示例21: create_test_file

​点赞 5

​

# 需要导入模块: import secrets [as 别名]

# 或者: from secrets import token_bytes [as 别名]

def create_test_file(self, file_path='.unittest_test_data', file_content=None, root_analysis=None):

"""Creates a test file and returns the path to the newly created file.

Any file created this way is automatically deleted after the test runs.

If file_path is relative then the file is created relative to SAQ_HOME.

If root_analysis is a RootAnalysis object then file_path is crated relative to the storage_dir of this analysis.

If file_content is not None then it is used as the content of the file.

Otherwise, 1024 random bytes are used."""

if not os.path.isabs(file_path):

if root_analysis:

target_file_path = os.path.join(root_analysis.storage_dir, file_path)

else:

target_file_path = abs_path(file_path)

mode = 'wb'

if isinstance(file_content, str):

mode = 'w'

with open(target_file_path, mode) as fp:

if file_content:

fp.write(file_content)

else:

fp.write(secrets.token_bytes(1024))

self.tracked_test_files.append(target_file_path)

return file_path

开发者ID:IntegralDefense，项目名称:ACE，代码行数:28，

示例22: __init__

​点赞 5

​

# 需要导入模块: import secrets [as 别名]

# 或者: from secrets import token_bytes [as 别名]

def __init__(self):

self._cookie = secrets.token_bytes(self.cookie_len)

开发者ID:twisteroidambassador，项目名称:ptadapter，代码行数:4，

示例23: authenticate

​点赞 5

​

# 需要导入模块: import secrets [as 别名]

# 或者: from secrets import token_bytes [as 别名]

def authenticate(

self,

reader: asyncio.StreamReader,

writer: asyncio.StreamWriter,

) -> bool:

"""(async) Authenticate a connecting client.

Returns:

True if authentication is successful and False otherwise. The

caller is responsible for closing the connection in case of

failure.

"""

writer.write(enums.ExtOrPortAuthTypes.SAFE_COOKIE

+ enums.ExtOrPortAuthTypes.END_AUTH_TYPES)

client_auth_type = await reader.readexactly(1)

if client_auth_type != enums.ExtOrPortAuthTypes.SAFE_COOKIE:

return False

client_nonce = await reader.readexactly(self.nonce_len)

server_nonce = secrets.token_bytes(self.nonce_len)

server_hash = self.hash(b''.join((

self.server_hash_header, client_nonce, server_nonce)))

writer.write(server_hash + server_nonce)

client_hash = await reader.readexactly(self.hash_len)

result = hmac.compare_digest(client_hash, self.hash(b''.join((

self.client_hash_header, client_nonce, server_nonce))))

writer.write(int(result).to_bytes(1, 'big'))

return result

开发者ID:twisteroidambassador，项目名称:ptadapter，代码行数:29，

示例24: __init__

​点赞 5

​

# 需要导入模块: import secrets [as 别名]

# 或者: from secrets import token_bytes [as 别名]

def __init__(self, client_socket, in_address):

self.auth_token = secrets.token_bytes(4)

self.session_unique = secrets.token_bytes(4)

self.client = client_socket

self.address = in_address

self.room_slot = -1

self.is_room_key = False

print("New session initialized")

if self.client is not None:

print("Session IP:", self.address[0], "Port:", self.address[1])

开发者ID:jglim，项目名称:gunbound-server，代码行数:12，

示例25: encrypt

​点赞 5

​

# 需要导入模块: import secrets [as 别名]

# 或者: from secrets import token_bytes [as 别名]

def encrypt(cls, value):

from django.conf import settings

salt = secrets.token_bytes(16)

key = cls.get_key(force_bytes(settings.SECRET_KEY), salt)

encrypted_value = b64d(Fernet(key).encrypt(value.encode()))

return b64e(b"%b%b" % (salt, encrypted_value)).decode()

开发者ID:stellar，项目名称:django-polaris，代码行数:9，

示例26: browse

​点赞 5

​

# 需要导入模块: import secrets [as 别名]

# 或者: from secrets import token_bytes [as 别名]

def browse(self, search_criteria, timeout=1, max=0):

key = secrets.token_bytes(16)

challenge = secrets.token_bytes(256)

self.send_browse_request(search_criteria, key, challenge)

return self.receive_browse_reply(timeout, max, key, challenge)

开发者ID:Kinnay，项目名称:NintendoClients，代码行数:7，

示例27: parse_challenge

​点赞 5

​

# 需要导入模块: import secrets [as 别名]

# 或者: from secrets import token_bytes [as 别名]

def parse_challenge(self, stream):

if self.settings.get("pia.lan_version") == 0:

return b""

version = stream.u8()

if version != self.settings.get("pia.lan_version"):

logger.warning("Browse request has unexpected version number")

return None

crypto = stream.bool()

nonce = stream.u64()

key = stream.read(16)

tag = stream.read(16)

challenge = stream.read(256)

if crypto:

challenge_key = self.generate_challenge_key(key)

nonce = self.generate_nonce(nonce)

aes = AES.new(challenge_key, AES.MODE_GCM, nonce=nonce)

try:

challenge = aes.decrypt_and_verify(challenge, tag)

except ValueError:

logger.warning("Challenge has incorrect authentication tag")

return None

new_key = secrets.token_bytes(16)

self.nonce_counter += 1

nonce = self.generate_nonce(self.nonce_counter)

reply = struct.pack(">Q", self.nonce_counter) + new_key

reply += self.generate_challenge_reply(nonce, new_key + key, challenge)

else:

reply = secrets.token_bytes(16 + 16 + 16)

return reply

开发者ID:Kinnay，项目名称:NintendoClients，代码行数:41，

示例28: create

​点赞 5

​

# 需要导入模块: import secrets [as 别名]

# 或者: from secrets import token_bytes [as 别名]

def create(self, settings):

session = MatchmakeSession()

session.session_key = secrets.token_bytes(16)

self.create_mesh(session.session_key, settings)

开发者ID:Kinnay，项目名称:NintendoClients，代码行数:6，

注：本文中的secrets.token_bytes方法示例整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。