Sadmin
f20080305d
Определение специфических рекомендаций (матрица профилей §8):
1. Ж/д слой (закрыт мёртвый railway ×2.5 у РАС):
- /api/v1/water/{case_id} отдаёт railway=rail как LineString
(без service/industrial/military веток), кэш общий v2;
- railway_warning «перекрыть/проверить немедленно» по профилям;
- SearchMap: Polyline слой ж/д (тёмно-красный), счётчики 💧/🚂.
2. cant_swim → профиль не_умеет_плавать (water ×3.0, без изменения
радиуса, critical_warning «обследовать водоёмы НЕМЕДЛЕННО»):
- раньше чекбокс влиял только на текст, в скоринге был пробел;
- derive в analyze._derive_profiles — работает и для closed_cases.
3. unmodeled_profiles: ДЦП/слабое зрение/слух — честная пометка
«вне поведенческой модели» с пояснением (vector_tasks B12:
профили без аналога не выдавать за учтённые); блок на фронте
в карточке здоровья.
Площадь воды: сферический эксцесс, проверен на квадрате 53° (744017 м²
vs 743272 точного). Тесты: 202 passed (новый test_cant_swim_profile).
143 lines
3.6 KiB
Python
143 lines
3.6 KiB
Python
import base64
|
|
import binascii
|
|
import re
|
|
from typing import Optional, Union
|
|
|
|
try:
|
|
from cryptography.hazmat.primitives.asymmetric.ec import EllipticCurve
|
|
from cryptography.hazmat.primitives.asymmetric.utils import (
|
|
decode_dss_signature,
|
|
encode_dss_signature,
|
|
)
|
|
except ModuleNotFoundError:
|
|
pass
|
|
|
|
|
|
def force_bytes(value: Union[bytes, str]) -> bytes:
|
|
if isinstance(value, str):
|
|
return value.encode("utf-8")
|
|
elif isinstance(value, bytes):
|
|
return value
|
|
else:
|
|
raise TypeError("Expected a string value")
|
|
|
|
|
|
def base64url_decode(input: Union[bytes, str]) -> bytes:
|
|
input_bytes = force_bytes(input)
|
|
|
|
rem = len(input_bytes) % 4
|
|
|
|
if rem > 0:
|
|
input_bytes += b"=" * (4 - rem)
|
|
|
|
return base64.urlsafe_b64decode(input_bytes)
|
|
|
|
|
|
def base64url_encode(input: bytes) -> bytes:
|
|
return base64.urlsafe_b64encode(input).replace(b"=", b"")
|
|
|
|
|
|
def to_base64url_uint(val: int, *, bit_length: Optional[int] = None) -> bytes:
|
|
if val < 0:
|
|
raise ValueError("Must be a positive integer")
|
|
|
|
int_bytes = bytes_from_int(val, bit_length=bit_length)
|
|
|
|
if len(int_bytes) == 0:
|
|
int_bytes = b"\x00"
|
|
|
|
return base64url_encode(int_bytes)
|
|
|
|
|
|
def from_base64url_uint(val: Union[bytes, str]) -> int:
|
|
data = base64url_decode(force_bytes(val))
|
|
return int.from_bytes(data, byteorder="big")
|
|
|
|
|
|
def number_to_bytes(num: int, num_bytes: int) -> bytes:
|
|
padded_hex = "%0*x" % (2 * num_bytes, num)
|
|
return binascii.a2b_hex(padded_hex.encode("ascii"))
|
|
|
|
|
|
def bytes_to_number(string: bytes) -> int:
|
|
return int(binascii.b2a_hex(string), 16)
|
|
|
|
|
|
def bytes_from_int(val: int, *, bit_length: Optional[int] = None) -> bytes:
|
|
if bit_length is None:
|
|
bit_length = val.bit_length()
|
|
byte_length = (bit_length + 7) // 8
|
|
|
|
return val.to_bytes(byte_length, "big", signed=False)
|
|
|
|
|
|
def der_to_raw_signature(der_sig: bytes, curve: "EllipticCurve") -> bytes:
|
|
num_bits = curve.key_size
|
|
num_bytes = (num_bits + 7) // 8
|
|
|
|
r, s = decode_dss_signature(der_sig)
|
|
|
|
return number_to_bytes(r, num_bytes) + number_to_bytes(s, num_bytes)
|
|
|
|
|
|
def raw_to_der_signature(raw_sig: bytes, curve: "EllipticCurve") -> bytes:
|
|
num_bits = curve.key_size
|
|
num_bytes = (num_bits + 7) // 8
|
|
|
|
if len(raw_sig) != 2 * num_bytes:
|
|
raise ValueError("Invalid signature")
|
|
|
|
r = bytes_to_number(raw_sig[:num_bytes])
|
|
s = bytes_to_number(raw_sig[num_bytes:])
|
|
|
|
return bytes(encode_dss_signature(r, s))
|
|
|
|
|
|
# Based on https://github.com/hynek/pem/blob/7ad94db26b0bc21d10953f5dbad3acfdfacf57aa/src/pem/_core.py#L224-L252
|
|
_PEMS = {
|
|
b"CERTIFICATE",
|
|
b"TRUSTED CERTIFICATE",
|
|
b"PRIVATE KEY",
|
|
b"PUBLIC KEY",
|
|
b"ENCRYPTED PRIVATE KEY",
|
|
b"OPENSSH PRIVATE KEY",
|
|
b"DSA PRIVATE KEY",
|
|
b"RSA PRIVATE KEY",
|
|
b"RSA PUBLIC KEY",
|
|
b"EC PRIVATE KEY",
|
|
b"DH PARAMETERS",
|
|
b"NEW CERTIFICATE REQUEST",
|
|
b"CERTIFICATE REQUEST",
|
|
b"SSH2 PUBLIC KEY",
|
|
b"SSH2 ENCRYPTED PRIVATE KEY",
|
|
b"X509 CRL",
|
|
}
|
|
|
|
_PEM_RE = re.compile(
|
|
b"----[- ]BEGIN ("
|
|
+ b"|".join(_PEMS)
|
|
+ b""")[- ]----\r?
|
|
.+?\r?
|
|
----[- ]END \\1[- ]----\r?\n?""",
|
|
re.DOTALL,
|
|
)
|
|
|
|
|
|
def is_pem_format(key: bytes) -> bool:
|
|
return bool(_PEM_RE.search(key))
|
|
|
|
|
|
# Based on https://github.com/pyca/cryptography/blob/bcb70852d577b3f490f015378c75cba74986297b/src/cryptography/hazmat/primitives/serialization/ssh.py#L40-L46
|
|
_SSH_KEY_FORMATS = (
|
|
b"ssh-ed25519",
|
|
b"ssh-rsa",
|
|
b"ssh-dss",
|
|
b"ecdsa-sha2-nistp256",
|
|
b"ecdsa-sha2-nistp384",
|
|
b"ecdsa-sha2-nistp521",
|
|
)
|
|
|
|
|
|
def is_ssh_key(key: bytes) -> bool:
|
|
return key.startswith(_SSH_KEY_FORMATS)
|