Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Identity tokens #1728

Merged
merged 3 commits into from
Oct 18, 2024
Merged

Identity tokens #1728

Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
612885a
Fix variable naming
cziebuhr Oct 16, 2024
e7f1397
Fix password decryption with Aes256Sha256RsaPss
cziebuhr Oct 11, 2024
33faf14
Fix specifying allowed UserIdentityTokens
cziebuhr Oct 11, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 8 additions & 8 deletions asyncua/common/connection.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,7 +84,7 @@ class MessageChunk:
Message Chunk, as described in OPC UA specs Part 6, 6.7.2.
"""

def __init__(self, security_policy, body=b'', msg_type=ua.MessageType.SecureMessage, chunk_type=ua.ChunkType.Single):
def __init__(self, crypto, body=b'', msg_type=ua.MessageType.SecureMessage, chunk_type=ua.ChunkType.Single):
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can't you add typing so we know what that object is?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not that easily right now. It could be security_policies.Cryptography or ua.CryptographyNone. I can address it in the next PR which will be ready soon.

self.MessageHeader = ua.Header(msg_type, chunk_type)
if msg_type in (ua.MessageType.SecureMessage, ua.MessageType.SecureClose):
self.SecurityHeader = ua.SymmetricAlgorithmHeader()
Expand All @@ -94,7 +94,7 @@ def __init__(self, security_policy, body=b'', msg_type=ua.MessageType.SecureMess
raise ua.UaError(f"Unsupported message type: {msg_type}")
self.SequenceHeader = ua.SequenceHeader()
self.Body = body
self.security_policy = security_policy
self.crypto = crypto

@staticmethod
def from_binary(security_policy, data):
Expand Down Expand Up @@ -134,20 +134,20 @@ def from_header_and_body(security_policy, header, buf, use_prev_key=False):
return obj

def encrypted_size(self, plain_size):
size = plain_size + self.security_policy.signature_size()
pbs = self.security_policy.plain_block_size()
size = plain_size + self.crypto.signature_size()
pbs = self.crypto.plain_block_size()
if size % pbs != 0:
raise ua.UaError("Encryption error")
return size // pbs * self.security_policy.encrypted_block_size()
return size // pbs * self.crypto.encrypted_block_size()

def to_binary(self):
security = struct_to_binary(self.SecurityHeader)
encrypted_part = struct_to_binary(self.SequenceHeader) + self.Body
encrypted_part += self.security_policy.padding(len(encrypted_part))
encrypted_part += self.crypto.padding(len(encrypted_part))
self.MessageHeader.body_size = len(security) + self.encrypted_size(len(encrypted_part))
header = header_to_binary(self.MessageHeader)
encrypted_part += self.security_policy.signature(header + security + encrypted_part)
return header + security + self.security_policy.encrypt(encrypted_part)
encrypted_part += self.crypto.signature(header + security + encrypted_part)
return header + security + self.crypto.encrypt(encrypted_part)

@staticmethod
def max_body_size(crypto, max_chunk_size):
Expand Down