mirror of
https://github.com/9001/copyparty.git
synced 2025-08-18 01:22:13 -06:00
print qr-code on startup
This commit is contained in:
parent
d284b46a3e
commit
dbb3edec77
|
@ -580,6 +580,15 @@ def run_argparse(argv: list[str], formatter: Any, retry: bool) -> argparse.Names
|
||||||
ap2.add_argument("--license", action="store_true", help="show licenses and exit")
|
ap2.add_argument("--license", action="store_true", help="show licenses and exit")
|
||||||
ap2.add_argument("--version", action="store_true", help="show versions and exit")
|
ap2.add_argument("--version", action="store_true", help="show versions and exit")
|
||||||
|
|
||||||
|
ap2 = ap.add_argument_group('qr options')
|
||||||
|
ap2.add_argument("--qr", action="store_true", help="show http:// QR-code on startup")
|
||||||
|
ap2.add_argument("--qrs", action="store_true", help="show https:// QR-code on startup")
|
||||||
|
ap2.add_argument("--qrl", metavar="PATH", type=u, default="", help="location to include in the url, for example [\033[32mpriv/?pw=hunter2\033[0m] (also enables --qr if not --qrs)")
|
||||||
|
ap2.add_argument("--qr-ip", metavar="PREFIX", type=u, default="", help="select IP which starts with PREFIX")
|
||||||
|
ap2.add_argument("--qr-fg", metavar="COLOR", type=int, default=16, help="foreground")
|
||||||
|
ap2.add_argument("--qr-bg", metavar="COLOR", type=int, default=229, help="background (white=255)")
|
||||||
|
ap2.add_argument("--qr-pad", metavar="CELLS", type=int, default=4, help="padding (spec says 4 or more)")
|
||||||
|
|
||||||
ap2 = ap.add_argument_group('upload options')
|
ap2 = ap.add_argument_group('upload options')
|
||||||
ap2.add_argument("--dotpart", action="store_true", help="dotfile incomplete uploads, hiding them from clients unless -ed")
|
ap2.add_argument("--dotpart", action="store_true", help="dotfile incomplete uploads, hiding them from clients unless -ed")
|
||||||
ap2.add_argument("--plain-ip", action="store_true", help="when avoiding filename collisions by appending the uploader's ip to the filename: append the plaintext ip instead of salting and hashing the ip")
|
ap2.add_argument("--plain-ip", action="store_true", help="when avoiding filename collisions by appending the uploader's ip to the filename: append the plaintext ip instead of salting and hashing the ip")
|
||||||
|
@ -824,8 +833,10 @@ def main(argv: Optional[list[str]] = None) -> None:
|
||||||
time.sleep(2)
|
time.sleep(2)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
if len(argv) == 1 and (ANYWIN or not os.geteuid()):
|
if len(argv) == 1:
|
||||||
argv.extend(["-p80,443,3923", "--ign-ebind"])
|
argv.extend(["--qr"])
|
||||||
|
if ANYWIN or not os.geteuid():
|
||||||
|
argv.extend(["-p80,443,3923", "--ign-ebind"])
|
||||||
except:
|
except:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
@ -908,6 +919,9 @@ def main(argv: Optional[list[str]] = None) -> None:
|
||||||
zs = "argument {} cannot be '{}'; try one of these: {}"
|
zs = "argument {} cannot be '{}'; try one of these: {}"
|
||||||
raise Exception(zs.format(arg, val, okays))
|
raise Exception(zs.format(arg, val, okays))
|
||||||
|
|
||||||
|
if al.qrl and not al.qrs:
|
||||||
|
al.qr = True
|
||||||
|
|
||||||
if HAVE_SSL:
|
if HAVE_SSL:
|
||||||
if al.ssl_ver:
|
if al.ssl_ver:
|
||||||
configure_ssl_ver(al)
|
configure_ssl_ver(al)
|
||||||
|
|
|
@ -9,6 +9,7 @@ import threading
|
||||||
|
|
||||||
import queue
|
import queue
|
||||||
|
|
||||||
|
from .__init__ import ANYWIN
|
||||||
from .authsrv import AuthSrv
|
from .authsrv import AuthSrv
|
||||||
from .broker_util import BrokerCli, ExceptionalQueue
|
from .broker_util import BrokerCli, ExceptionalQueue
|
||||||
from .httpsrv import HttpSrv
|
from .httpsrv import HttpSrv
|
||||||
|
@ -48,7 +49,11 @@ class MpWorker(BrokerCli):
|
||||||
# we inherited signal_handler from parent,
|
# we inherited signal_handler from parent,
|
||||||
# replace it with something harmless
|
# replace it with something harmless
|
||||||
if not FAKE_MP:
|
if not FAKE_MP:
|
||||||
for sig in [signal.SIGINT, signal.SIGTERM, signal.SIGUSR1]:
|
sigs = [signal.SIGINT, signal.SIGTERM]
|
||||||
|
if not ANYWIN:
|
||||||
|
sigs.append(signal.SIGUSR1)
|
||||||
|
|
||||||
|
for sig in sigs:
|
||||||
signal.signal(sig, self.signal_handler)
|
signal.signal(sig, self.signal_handler)
|
||||||
|
|
||||||
# starting to look like a good idea
|
# starting to look like a good idea
|
||||||
|
|
|
@ -81,8 +81,8 @@ from .util import (
|
||||||
)
|
)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
from typing import Any, Generator, Match, Optional, Pattern, Type, Union
|
|
||||||
import typing
|
import typing
|
||||||
|
from typing import Any, Generator, Match, Optional, Pattern, Type, Union
|
||||||
except:
|
except:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
586
copyparty/stolen/qrcodegen.py
Normal file
586
copyparty/stolen/qrcodegen.py
Normal file
|
@ -0,0 +1,586 @@
|
||||||
|
# coding: utf-8
|
||||||
|
|
||||||
|
# modified copy of Project Nayuki's qrcodegen (MIT-licensed);
|
||||||
|
# https://github.com/nayuki/QR-Code-generator/blob/daa3114/python/qrcodegen.py
|
||||||
|
# the original ^ is extremely well commented so refer to that for explanations
|
||||||
|
|
||||||
|
# hacks: binary-only, auto-ecc, render, py2-compat
|
||||||
|
|
||||||
|
from __future__ import print_function, unicode_literals
|
||||||
|
|
||||||
|
import collections
|
||||||
|
import itertools
|
||||||
|
|
||||||
|
try:
|
||||||
|
from collections.abc import Sequence
|
||||||
|
|
||||||
|
from typing import Callable, List, Optional, Tuple, Union
|
||||||
|
except:
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
def num_char_count_bits(ver: int) -> int:
|
||||||
|
return 16 if (ver + 7) // 17 else 8
|
||||||
|
|
||||||
|
|
||||||
|
class Ecc(object):
|
||||||
|
ordinal: int
|
||||||
|
formatbits: int
|
||||||
|
|
||||||
|
def __init__(self, i: int, fb: int) -> None:
|
||||||
|
self.ordinal = i
|
||||||
|
self.formatbits = fb
|
||||||
|
|
||||||
|
LOW: "Ecc"
|
||||||
|
MEDIUM: "Ecc"
|
||||||
|
QUARTILE: "Ecc"
|
||||||
|
HIGH: "Ecc"
|
||||||
|
|
||||||
|
|
||||||
|
Ecc.LOW = Ecc(0, 1)
|
||||||
|
Ecc.MEDIUM = Ecc(1, 0)
|
||||||
|
Ecc.QUARTILE = Ecc(2, 3)
|
||||||
|
Ecc.HIGH = Ecc(3, 2)
|
||||||
|
|
||||||
|
|
||||||
|
class QrSegment(object):
|
||||||
|
@staticmethod
|
||||||
|
def make_seg(data: Union[bytes, Sequence[int]]) -> "QrSegment":
|
||||||
|
bb = _BitBuffer()
|
||||||
|
for b in data:
|
||||||
|
bb.append_bits(b, 8)
|
||||||
|
return QrSegment(len(data), bb)
|
||||||
|
|
||||||
|
numchars: int # num bytes, not the same as the data's bit length
|
||||||
|
bitdata: List[int] # The data bits of this segment
|
||||||
|
|
||||||
|
def __init__(self, numch: int, bitdata: Sequence[int]) -> None:
|
||||||
|
if numch < 0:
|
||||||
|
raise ValueError()
|
||||||
|
self.numchars = numch
|
||||||
|
self.bitdata = list(bitdata)
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def get_total_bits(segs: Sequence["QrSegment"], ver: int) -> Optional[int]:
|
||||||
|
result = 0
|
||||||
|
for seg in segs:
|
||||||
|
ccbits: int = num_char_count_bits(ver)
|
||||||
|
if seg.numchars >= (1 << ccbits):
|
||||||
|
return None # segment length doesn't fit the field's bit width
|
||||||
|
result += 4 + ccbits + len(seg.bitdata)
|
||||||
|
return result
|
||||||
|
|
||||||
|
|
||||||
|
class QrCode(object):
|
||||||
|
@staticmethod
|
||||||
|
def encode_binary(data: Union[bytes, Sequence[int]]) -> "QrCode":
|
||||||
|
return QrCode.encode_segments([QrSegment.make_seg(data)])
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def encode_segments(
|
||||||
|
segs: Sequence[QrSegment],
|
||||||
|
ecl: Ecc = Ecc.LOW,
|
||||||
|
minver: int = 2,
|
||||||
|
maxver: int = 40,
|
||||||
|
mask: int = -1,
|
||||||
|
) -> "QrCode":
|
||||||
|
for ver in range(minver, maxver + 1):
|
||||||
|
datacapacitybits: int = QrCode._get_num_data_codewords(ver, ecl) * 8
|
||||||
|
datausedbits: Optional[int] = QrSegment.get_total_bits(segs, ver)
|
||||||
|
if (datausedbits is not None) and (datausedbits <= datacapacitybits):
|
||||||
|
break
|
||||||
|
|
||||||
|
assert datausedbits
|
||||||
|
|
||||||
|
for newecl in (
|
||||||
|
Ecc.MEDIUM,
|
||||||
|
Ecc.QUARTILE,
|
||||||
|
Ecc.HIGH,
|
||||||
|
):
|
||||||
|
if datausedbits <= QrCode._get_num_data_codewords(ver, newecl) * 8:
|
||||||
|
ecl = newecl
|
||||||
|
|
||||||
|
# Concatenate all segments to create the data bit string
|
||||||
|
bb = _BitBuffer()
|
||||||
|
for seg in segs:
|
||||||
|
bb.append_bits(4, 4)
|
||||||
|
bb.append_bits(seg.numchars, num_char_count_bits(ver))
|
||||||
|
bb.extend(seg.bitdata)
|
||||||
|
assert len(bb) == datausedbits
|
||||||
|
|
||||||
|
# Add terminator and pad up to a byte if applicable
|
||||||
|
datacapacitybits = QrCode._get_num_data_codewords(ver, ecl) * 8
|
||||||
|
assert len(bb) <= datacapacitybits
|
||||||
|
bb.append_bits(0, min(4, datacapacitybits - len(bb)))
|
||||||
|
bb.append_bits(0, -len(bb) % 8)
|
||||||
|
assert len(bb) % 8 == 0
|
||||||
|
|
||||||
|
# Pad with alternating bytes until data capacity is reached
|
||||||
|
for padbyte in itertools.cycle((0xEC, 0x11)):
|
||||||
|
if len(bb) >= datacapacitybits:
|
||||||
|
break
|
||||||
|
bb.append_bits(padbyte, 8)
|
||||||
|
|
||||||
|
# Pack bits into bytes in big endian
|
||||||
|
datacodewords = bytearray([0] * (len(bb) // 8))
|
||||||
|
for (i, bit) in enumerate(bb):
|
||||||
|
datacodewords[i >> 3] |= bit << (7 - (i & 7))
|
||||||
|
|
||||||
|
return QrCode(ver, ecl, datacodewords, mask)
|
||||||
|
|
||||||
|
ver: int
|
||||||
|
size: int # w/h; 21..177 (ver * 4 + 17)
|
||||||
|
ecclvl: Ecc
|
||||||
|
mask: int # 0..7
|
||||||
|
modules: List[List[bool]]
|
||||||
|
unmaskable: List[List[bool]]
|
||||||
|
|
||||||
|
def __init__(
|
||||||
|
self,
|
||||||
|
ver: int,
|
||||||
|
ecclvl: Ecc,
|
||||||
|
datacodewords: Union[bytes, Sequence[int]],
|
||||||
|
msk: int,
|
||||||
|
) -> None:
|
||||||
|
self.ver = ver
|
||||||
|
self.size = ver * 4 + 17
|
||||||
|
self.ecclvl = ecclvl
|
||||||
|
|
||||||
|
self.modules = [[False] * self.size for _ in range(self.size)]
|
||||||
|
self.unmaskable = [[False] * self.size for _ in range(self.size)]
|
||||||
|
|
||||||
|
# Compute ECC, draw modules
|
||||||
|
self._draw_function_patterns()
|
||||||
|
allcodewords: bytes = self._add_ecc_and_interleave(bytearray(datacodewords))
|
||||||
|
self._draw_codewords(allcodewords)
|
||||||
|
|
||||||
|
if msk == -1: # automask
|
||||||
|
minpenalty: int = 1 << 32
|
||||||
|
for i in range(8):
|
||||||
|
self._apply_mask(i)
|
||||||
|
self._draw_format_bits(i)
|
||||||
|
penalty = self._get_penalty_score()
|
||||||
|
if penalty < minpenalty:
|
||||||
|
msk = i
|
||||||
|
minpenalty = penalty
|
||||||
|
self._apply_mask(i) # xor/undo
|
||||||
|
|
||||||
|
assert 0 <= msk <= 7
|
||||||
|
self.mask = msk
|
||||||
|
self._apply_mask(msk) # Apply the final choice of mask
|
||||||
|
self._draw_format_bits(msk) # Overwrite old format bits
|
||||||
|
|
||||||
|
def render(self, pad=4) -> str:
|
||||||
|
tab = self.modules
|
||||||
|
sz = self.size
|
||||||
|
if sz % 2:
|
||||||
|
tab.append([False] * sz)
|
||||||
|
|
||||||
|
tab = [[False] * sz] * pad + tab + [[False] * sz] * pad
|
||||||
|
tab = [[False] * pad + x + [False] * pad for x in tab]
|
||||||
|
|
||||||
|
rows: list[str] = []
|
||||||
|
for y in range(0, len(tab), 2):
|
||||||
|
row = ""
|
||||||
|
for x in range(len(tab[y])):
|
||||||
|
v = 2 if tab[y][x] else 0
|
||||||
|
v += 1 if tab[y + 1][x] else 0
|
||||||
|
row += " ▄▀█"[v]
|
||||||
|
rows.append(row)
|
||||||
|
|
||||||
|
return "\n".join(rows)
|
||||||
|
|
||||||
|
def _draw_function_patterns(self) -> None:
|
||||||
|
# Draw horizontal and vertical timing patterns
|
||||||
|
for i in range(self.size):
|
||||||
|
self._set_function_module(6, i, i % 2 == 0)
|
||||||
|
self._set_function_module(i, 6, i % 2 == 0)
|
||||||
|
|
||||||
|
# Draw 3 finder patterns (all corners except bottom right; overwrites some timing modules)
|
||||||
|
self._draw_finder_pattern(3, 3)
|
||||||
|
self._draw_finder_pattern(self.size - 4, 3)
|
||||||
|
self._draw_finder_pattern(3, self.size - 4)
|
||||||
|
|
||||||
|
# Draw numerous alignment patterns
|
||||||
|
alignpatpos: List[int] = self._get_alignment_pattern_positions()
|
||||||
|
numalign: int = len(alignpatpos)
|
||||||
|
skips: Sequence[Tuple[int, int]] = (
|
||||||
|
(0, 0),
|
||||||
|
(0, numalign - 1),
|
||||||
|
(numalign - 1, 0),
|
||||||
|
)
|
||||||
|
for i in range(numalign):
|
||||||
|
for j in range(numalign):
|
||||||
|
if (i, j) not in skips: # avoid finder corners
|
||||||
|
self._draw_alignment_pattern(alignpatpos[i], alignpatpos[j])
|
||||||
|
|
||||||
|
# draw config data with dummy mask value; ctor overwrites it
|
||||||
|
self._draw_format_bits(0)
|
||||||
|
self._draw_ver()
|
||||||
|
|
||||||
|
def _draw_format_bits(self, mask: int) -> None:
|
||||||
|
# Calculate error correction code and pack bits; ecclvl is uint2, mask is uint3
|
||||||
|
data: int = self.ecclvl.formatbits << 3 | mask
|
||||||
|
rem: int = data
|
||||||
|
for _ in range(10):
|
||||||
|
rem = (rem << 1) ^ ((rem >> 9) * 0x537)
|
||||||
|
bits: int = (data << 10 | rem) ^ 0x5412 # uint15
|
||||||
|
assert bits >> 15 == 0
|
||||||
|
|
||||||
|
# first copy
|
||||||
|
for i in range(0, 6):
|
||||||
|
self._set_function_module(8, i, _get_bit(bits, i))
|
||||||
|
self._set_function_module(8, 7, _get_bit(bits, 6))
|
||||||
|
self._set_function_module(8, 8, _get_bit(bits, 7))
|
||||||
|
self._set_function_module(7, 8, _get_bit(bits, 8))
|
||||||
|
for i in range(9, 15):
|
||||||
|
self._set_function_module(14 - i, 8, _get_bit(bits, i))
|
||||||
|
|
||||||
|
# second copy
|
||||||
|
for i in range(0, 8):
|
||||||
|
self._set_function_module(self.size - 1 - i, 8, _get_bit(bits, i))
|
||||||
|
for i in range(8, 15):
|
||||||
|
self._set_function_module(8, self.size - 15 + i, _get_bit(bits, i))
|
||||||
|
self._set_function_module(8, self.size - 8, True) # Always dark
|
||||||
|
|
||||||
|
def _draw_ver(self) -> None:
|
||||||
|
if self.ver < 7:
|
||||||
|
return
|
||||||
|
|
||||||
|
# Calculate error correction code and pack bits
|
||||||
|
rem: int = self.ver # ver is uint6, 7..40
|
||||||
|
for _ in range(12):
|
||||||
|
rem = (rem << 1) ^ ((rem >> 11) * 0x1F25)
|
||||||
|
bits: int = self.ver << 12 | rem # uint18
|
||||||
|
assert bits >> 18 == 0
|
||||||
|
|
||||||
|
# Draw two copies
|
||||||
|
for i in range(18):
|
||||||
|
bit: bool = _get_bit(bits, i)
|
||||||
|
a: int = self.size - 11 + i % 3
|
||||||
|
b: int = i // 3
|
||||||
|
self._set_function_module(a, b, bit)
|
||||||
|
self._set_function_module(b, a, bit)
|
||||||
|
|
||||||
|
def _draw_finder_pattern(self, x: int, y: int) -> None:
|
||||||
|
for dy in range(-4, 5):
|
||||||
|
for dx in range(-4, 5):
|
||||||
|
xx, yy = x + dx, y + dy
|
||||||
|
if (0 <= xx < self.size) and (0 <= yy < self.size):
|
||||||
|
# Chebyshev/infinity norm
|
||||||
|
self._set_function_module(
|
||||||
|
xx, yy, max(abs(dx), abs(dy)) not in (2, 4)
|
||||||
|
)
|
||||||
|
|
||||||
|
def _draw_alignment_pattern(self, x: int, y: int) -> None:
|
||||||
|
for dy in range(-2, 3):
|
||||||
|
for dx in range(-2, 3):
|
||||||
|
self._set_function_module(x + dx, y + dy, max(abs(dx), abs(dy)) != 1)
|
||||||
|
|
||||||
|
def _set_function_module(self, x: int, y: int, isdark: bool) -> None:
|
||||||
|
self.modules[y][x] = isdark
|
||||||
|
self.unmaskable[y][x] = True
|
||||||
|
|
||||||
|
def _add_ecc_and_interleave(self, data: bytearray) -> bytes:
|
||||||
|
ver: int = self.ver
|
||||||
|
assert len(data) == QrCode._get_num_data_codewords(ver, self.ecclvl)
|
||||||
|
|
||||||
|
# Calculate parameter numbers
|
||||||
|
numblocks: int = QrCode._NUM_ERROR_CORRECTION_BLOCKS[self.ecclvl.ordinal][ver]
|
||||||
|
blockecclen: int = QrCode._ECC_CODEWORDS_PER_BLOCK[self.ecclvl.ordinal][ver]
|
||||||
|
rawcodewords: int = QrCode._get_num_raw_data_modules(ver) // 8
|
||||||
|
numshortblocks: int = numblocks - rawcodewords % numblocks
|
||||||
|
shortblocklen: int = rawcodewords // numblocks
|
||||||
|
|
||||||
|
# Split data into blocks and append ECC to each block
|
||||||
|
blocks: List[bytes] = []
|
||||||
|
rsdiv: bytes = QrCode._reed_solomon_compute_divisor(blockecclen)
|
||||||
|
k: int = 0
|
||||||
|
for i in range(numblocks):
|
||||||
|
dat: bytearray = data[
|
||||||
|
k : k + shortblocklen - blockecclen + (0 if i < numshortblocks else 1)
|
||||||
|
]
|
||||||
|
k += len(dat)
|
||||||
|
ecc: bytes = QrCode._reed_solomon_compute_remainder(dat, rsdiv)
|
||||||
|
if i < numshortblocks:
|
||||||
|
dat.append(0)
|
||||||
|
blocks.append(dat + ecc)
|
||||||
|
assert k == len(data)
|
||||||
|
|
||||||
|
# Interleave (not concatenate) the bytes from every block into a single sequence
|
||||||
|
result = bytearray()
|
||||||
|
for i in range(len(blocks[0])):
|
||||||
|
for (j, blk) in enumerate(blocks):
|
||||||
|
# Skip the padding byte in short blocks
|
||||||
|
if (i != shortblocklen - blockecclen) or (j >= numshortblocks):
|
||||||
|
result.append(blk[i])
|
||||||
|
assert len(result) == rawcodewords
|
||||||
|
return result
|
||||||
|
|
||||||
|
def _draw_codewords(self, data: bytes) -> None:
|
||||||
|
assert len(data) == QrCode._get_num_raw_data_modules(self.ver) // 8
|
||||||
|
|
||||||
|
i: int = 0 # Bit index into the data
|
||||||
|
for right in range(self.size - 1, 0, -2):
|
||||||
|
# idx of right column in each column pair
|
||||||
|
if right <= 6:
|
||||||
|
right -= 1
|
||||||
|
for vert in range(self.size): # Vertical counter
|
||||||
|
for j in range(2):
|
||||||
|
x: int = right - j
|
||||||
|
upward: bool = (right + 1) & 2 == 0
|
||||||
|
y: int = (self.size - 1 - vert) if upward else vert
|
||||||
|
if (not self.unmaskable[y][x]) and (i < len(data) * 8):
|
||||||
|
self.modules[y][x] = _get_bit(data[i >> 3], 7 - (i & 7))
|
||||||
|
i += 1
|
||||||
|
# any remainder bits (0..7) were set 0/false/light by ctor
|
||||||
|
|
||||||
|
assert i == len(data) * 8
|
||||||
|
|
||||||
|
def _apply_mask(self, mask: int) -> None:
|
||||||
|
masker: Callable[[int, int], int] = QrCode._MASK_PATTERNS[mask]
|
||||||
|
for y in range(self.size):
|
||||||
|
for x in range(self.size):
|
||||||
|
self.modules[y][x] ^= (masker(x, y) == 0) and (
|
||||||
|
not self.unmaskable[y][x]
|
||||||
|
)
|
||||||
|
|
||||||
|
def _get_penalty_score(self) -> int:
|
||||||
|
result: int = 0
|
||||||
|
size: int = self.size
|
||||||
|
modules: List[List[bool]] = self.modules
|
||||||
|
|
||||||
|
# Adjacent modules in row having same color, and finder-like patterns
|
||||||
|
for y in range(size):
|
||||||
|
runcolor: bool = False
|
||||||
|
runx: int = 0
|
||||||
|
runhistory = collections.deque([0] * 7, 7)
|
||||||
|
for x in range(size):
|
||||||
|
if modules[y][x] == runcolor:
|
||||||
|
runx += 1
|
||||||
|
if runx == 5:
|
||||||
|
result += QrCode._PENALTY_N1
|
||||||
|
elif runx > 5:
|
||||||
|
result += 1
|
||||||
|
else:
|
||||||
|
self._finder_penalty_add_history(runx, runhistory)
|
||||||
|
if not runcolor:
|
||||||
|
result += (
|
||||||
|
self._finder_penalty_count_patterns(runhistory)
|
||||||
|
* QrCode._PENALTY_N3
|
||||||
|
)
|
||||||
|
runcolor = modules[y][x]
|
||||||
|
runx = 1
|
||||||
|
result += (
|
||||||
|
self._finder_penalty_terminate_and_count(runcolor, runx, runhistory)
|
||||||
|
* QrCode._PENALTY_N3
|
||||||
|
)
|
||||||
|
|
||||||
|
# Adjacent modules in column having same color, and finder-like patterns
|
||||||
|
for x in range(size):
|
||||||
|
runcolor = False
|
||||||
|
runy = 0
|
||||||
|
runhistory = collections.deque([0] * 7, 7)
|
||||||
|
for y in range(size):
|
||||||
|
if modules[y][x] == runcolor:
|
||||||
|
runy += 1
|
||||||
|
if runy == 5:
|
||||||
|
result += QrCode._PENALTY_N1
|
||||||
|
elif runy > 5:
|
||||||
|
result += 1
|
||||||
|
else:
|
||||||
|
self._finder_penalty_add_history(runy, runhistory)
|
||||||
|
if not runcolor:
|
||||||
|
result += (
|
||||||
|
self._finder_penalty_count_patterns(runhistory)
|
||||||
|
* QrCode._PENALTY_N3
|
||||||
|
)
|
||||||
|
runcolor = modules[y][x]
|
||||||
|
runy = 1
|
||||||
|
result += (
|
||||||
|
self._finder_penalty_terminate_and_count(runcolor, runy, runhistory)
|
||||||
|
* QrCode._PENALTY_N3
|
||||||
|
)
|
||||||
|
|
||||||
|
# 2*2 blocks of modules having same color
|
||||||
|
for y in range(size - 1):
|
||||||
|
for x in range(size - 1):
|
||||||
|
if (
|
||||||
|
modules[y][x]
|
||||||
|
== modules[y][x + 1]
|
||||||
|
== modules[y + 1][x]
|
||||||
|
== modules[y + 1][x + 1]
|
||||||
|
):
|
||||||
|
result += QrCode._PENALTY_N2
|
||||||
|
|
||||||
|
# Balance of dark and light modules
|
||||||
|
dark: int = sum((1 if cell else 0) for row in modules for cell in row)
|
||||||
|
total: int = size ** 2 # Note that size is odd, so dark/total != 1/2
|
||||||
|
|
||||||
|
# Compute the smallest integer k >= 0 such that (45-5k)% <= dark/total <= (55+5k)%
|
||||||
|
k: int = (abs(dark * 20 - total * 10) + total - 1) // total - 1
|
||||||
|
assert 0 <= k <= 9
|
||||||
|
result += k * QrCode._PENALTY_N4
|
||||||
|
assert 0 <= result <= 2568888
|
||||||
|
# ^ Non-tight upper bound based on default values of PENALTY_N1, ..., N4
|
||||||
|
|
||||||
|
return result
|
||||||
|
|
||||||
|
def _get_alignment_pattern_positions(self) -> List[int]:
|
||||||
|
ver: int = self.ver
|
||||||
|
if ver == 1:
|
||||||
|
return []
|
||||||
|
|
||||||
|
numalign: int = ver // 7 + 2
|
||||||
|
step: int = (
|
||||||
|
26
|
||||||
|
if (ver == 32)
|
||||||
|
else (ver * 4 + numalign * 2 + 1) // (numalign * 2 - 2) * 2
|
||||||
|
)
|
||||||
|
result: List[int] = [
|
||||||
|
(self.size - 7 - i * step) for i in range(numalign - 1)
|
||||||
|
] + [6]
|
||||||
|
return list(reversed(result))
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _get_num_raw_data_modules(ver: int) -> int:
|
||||||
|
result: int = (16 * ver + 128) * ver + 64
|
||||||
|
if ver >= 2:
|
||||||
|
numalign: int = ver // 7 + 2
|
||||||
|
result -= (25 * numalign - 10) * numalign - 55
|
||||||
|
if ver >= 7:
|
||||||
|
result -= 36
|
||||||
|
assert 208 <= result <= 29648
|
||||||
|
return result
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _get_num_data_codewords(ver: int, ecl: Ecc) -> int:
|
||||||
|
return (
|
||||||
|
QrCode._get_num_raw_data_modules(ver) // 8
|
||||||
|
- QrCode._ECC_CODEWORDS_PER_BLOCK[ecl.ordinal][ver]
|
||||||
|
* QrCode._NUM_ERROR_CORRECTION_BLOCKS[ecl.ordinal][ver]
|
||||||
|
)
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _reed_solomon_compute_divisor(degree: int) -> bytes:
|
||||||
|
if not (1 <= degree <= 255):
|
||||||
|
raise ValueError("Degree out of range")
|
||||||
|
|
||||||
|
# Polynomial coefficients are stored from highest to lowest power, excluding the leading term which is always 1.
|
||||||
|
# For example the polynomial x^3 + 255x^2 + 8x + 93 is stored as the uint8 array [255, 8, 93].
|
||||||
|
result = bytearray([0] * (degree - 1) + [1]) # start with monomial x^0
|
||||||
|
|
||||||
|
# Compute the product polynomial (x - r^0) * (x - r^1) * (x - r^2) * ... * (x - r^{degree-1}),
|
||||||
|
# and drop the highest monomial term which is always 1x^degree.
|
||||||
|
# Note that r = 0x02, which is a generator element of this field GF(2^8/0x11D).
|
||||||
|
root: int = 1
|
||||||
|
for _ in range(degree):
|
||||||
|
# Multiply the current product by (x - r^i)
|
||||||
|
for j in range(degree):
|
||||||
|
result[j] = QrCode._reed_solomon_multiply(result[j], root)
|
||||||
|
if j + 1 < degree:
|
||||||
|
result[j] ^= result[j + 1]
|
||||||
|
root = QrCode._reed_solomon_multiply(root, 0x02)
|
||||||
|
|
||||||
|
return result
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _reed_solomon_compute_remainder(data: bytes, divisor: bytes) -> bytes:
|
||||||
|
result = bytearray([0] * len(divisor))
|
||||||
|
for b in data: # Polynomial division
|
||||||
|
factor: int = b ^ result.pop(0)
|
||||||
|
result.append(0)
|
||||||
|
for (i, coef) in enumerate(divisor):
|
||||||
|
result[i] ^= QrCode._reed_solomon_multiply(coef, factor)
|
||||||
|
|
||||||
|
return result
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _reed_solomon_multiply(x: int, y: int) -> int:
|
||||||
|
if (x >> 8 != 0) or (y >> 8 != 0):
|
||||||
|
raise ValueError("Byte out of range")
|
||||||
|
z: int = 0 # Russian peasant multiplication
|
||||||
|
for i in reversed(range(8)):
|
||||||
|
z = (z << 1) ^ ((z >> 7) * 0x11D)
|
||||||
|
z ^= ((y >> i) & 1) * x
|
||||||
|
assert z >> 8 == 0
|
||||||
|
return z
|
||||||
|
|
||||||
|
def _finder_penalty_count_patterns(self, runhistory: collections.deque[int]) -> int:
|
||||||
|
n: int = runhistory[1]
|
||||||
|
assert n <= self.size * 3
|
||||||
|
core: bool = (
|
||||||
|
n > 0
|
||||||
|
and (runhistory[2] == runhistory[4] == runhistory[5] == n)
|
||||||
|
and runhistory[3] == n * 3
|
||||||
|
)
|
||||||
|
return (
|
||||||
|
1 if (core and runhistory[0] >= n * 4 and runhistory[6] >= n) else 0
|
||||||
|
) + (1 if (core and runhistory[6] >= n * 4 and runhistory[0] >= n) else 0)
|
||||||
|
|
||||||
|
def _finder_penalty_terminate_and_count(
|
||||||
|
self,
|
||||||
|
currentruncolor: bool,
|
||||||
|
currentrunlength: int,
|
||||||
|
runhistory: collections.deque[int],
|
||||||
|
) -> int:
|
||||||
|
if currentruncolor: # Terminate dark run
|
||||||
|
self._finder_penalty_add_history(currentrunlength, runhistory)
|
||||||
|
currentrunlength = 0
|
||||||
|
currentrunlength += self.size # Add light border to final run
|
||||||
|
self._finder_penalty_add_history(currentrunlength, runhistory)
|
||||||
|
return self._finder_penalty_count_patterns(runhistory)
|
||||||
|
|
||||||
|
def _finder_penalty_add_history(
|
||||||
|
self, currentrunlength: int, runhistory: collections.deque[int]
|
||||||
|
) -> None:
|
||||||
|
if runhistory[0] == 0:
|
||||||
|
currentrunlength += self.size # Add light border to initial run
|
||||||
|
|
||||||
|
runhistory.appendleft(currentrunlength)
|
||||||
|
|
||||||
|
_PENALTY_N1: int = 3
|
||||||
|
_PENALTY_N2: int = 3
|
||||||
|
_PENALTY_N3: int = 40
|
||||||
|
_PENALTY_N4: int = 10
|
||||||
|
|
||||||
|
# fmt: off
|
||||||
|
_ECC_CODEWORDS_PER_BLOCK: Sequence[Sequence[int]] = (
|
||||||
|
(-1, 7, 10, 15, 20, 26, 18, 20, 24, 30, 18, 20, 24, 26, 30, 22, 24, 28, 30, 28, 28, 28, 28, 30, 30, 26, 28, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30), # noqa: E241 # L
|
||||||
|
(-1, 10, 16, 26, 18, 24, 16, 18, 22, 22, 26, 30, 22, 22, 24, 24, 28, 28, 26, 26, 26, 26, 28, 28, 28, 28, 28, 28, 28, 28, 28, 28, 28, 28, 28, 28, 28, 28, 28, 28, 28), # noqa: E241 # M
|
||||||
|
(-1, 13, 22, 18, 26, 18, 24, 18, 22, 20, 24, 28, 26, 24, 20, 30, 24, 28, 28, 26, 30, 28, 30, 30, 30, 30, 28, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30), # noqa: E241 # Q
|
||||||
|
(-1, 17, 28, 22, 16, 22, 28, 26, 26, 24, 28, 24, 28, 22, 24, 24, 30, 28, 28, 26, 28, 30, 24, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30, 30)) # noqa: E241 # H
|
||||||
|
|
||||||
|
_NUM_ERROR_CORRECTION_BLOCKS: Sequence[Sequence[int]] = (
|
||||||
|
(-1, 1, 1, 1, 1, 1, 2, 2, 2, 2, 4, 4, 4, 4, 4, 6, 6, 6, 6, 7, 8, 8, 9, 9, 10, 12, 12, 12, 13, 14, 15, 16, 17, 18, 19, 19, 20, 21, 22, 24, 25), # noqa: E241 # L
|
||||||
|
(-1, 1, 1, 1, 2, 2, 4, 4, 4, 5, 5, 5, 8, 9, 9, 10, 10, 11, 13, 14, 16, 17, 17, 18, 20, 21, 23, 25, 26, 28, 29, 31, 33, 35, 37, 38, 40, 43, 45, 47, 49), # noqa: E241 # M
|
||||||
|
(-1, 1, 1, 2, 2, 4, 4, 6, 6, 8, 8, 8, 10, 12, 16, 12, 17, 16, 18, 21, 20, 23, 23, 25, 27, 29, 34, 34, 35, 38, 40, 43, 45, 48, 51, 53, 56, 59, 62, 65, 68), # noqa: E241 # Q
|
||||||
|
(-1, 1, 1, 2, 4, 4, 4, 5, 6, 8, 8, 11, 11, 16, 16, 18, 16, 19, 21, 25, 25, 25, 34, 30, 32, 35, 37, 40, 42, 45, 48, 51, 54, 57, 60, 63, 66, 70, 74, 77, 81)) # noqa: E241 # H
|
||||||
|
# fmt: on
|
||||||
|
|
||||||
|
_MASK_PATTERNS: Sequence[Callable[[int, int], int]] = (
|
||||||
|
(lambda x, y: (x + y) % 2),
|
||||||
|
(lambda x, y: y % 2),
|
||||||
|
(lambda x, y: x % 3),
|
||||||
|
(lambda x, y: (x + y) % 3),
|
||||||
|
(lambda x, y: (x // 3 + y // 2) % 2),
|
||||||
|
(lambda x, y: x * y % 2 + x * y % 3),
|
||||||
|
(lambda x, y: (x * y % 2 + x * y % 3) % 2),
|
||||||
|
(lambda x, y: ((x + y) % 2 + x * y % 3) % 2),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class _BitBuffer(list): # type: ignore
|
||||||
|
def append_bits(self, val: int, n: int) -> None:
|
||||||
|
if (n < 0) or (val >> n != 0):
|
||||||
|
raise ValueError("Value out of range")
|
||||||
|
|
||||||
|
self.extend(((val >> i) & 1) for i in reversed(range(n)))
|
||||||
|
|
||||||
|
|
||||||
|
def _get_bit(x: int, i: int) -> bool:
|
||||||
|
return (x >> i) & 1 != 0
|
||||||
|
|
||||||
|
|
||||||
|
class DataTooLongError(ValueError):
|
||||||
|
pass
|
|
@ -222,7 +222,11 @@ class SvcHub(object):
|
||||||
return
|
return
|
||||||
|
|
||||||
time.sleep(0.1) # purely cosmetic dw
|
time.sleep(0.1) # purely cosmetic dw
|
||||||
self.log("root", "workers OK\n")
|
if self.tcpsrv.qr:
|
||||||
|
self.log("qr-code", self.tcpsrv.qr)
|
||||||
|
else:
|
||||||
|
self.log("root", "workers OK\n")
|
||||||
|
|
||||||
self.up2k.init_vols()
|
self.up2k.init_vols()
|
||||||
|
|
||||||
thr = threading.Thread(target=self.sd_notify, name="sd-notify")
|
thr = threading.Thread(target=self.sd_notify, name="sd-notify")
|
||||||
|
@ -269,7 +273,8 @@ class SvcHub(object):
|
||||||
|
|
||||||
msg = "[+] opened logfile [{}]\n".format(fn)
|
msg = "[+] opened logfile [{}]\n".format(fn)
|
||||||
printed += msg
|
printed += msg
|
||||||
lh.write("t0: {:.3f}\nargv: {}\n\n{}".format(self.E.t0, " ".join(argv), printed))
|
t = "t0: {:.3f}\nargv: {}\n\n{}"
|
||||||
|
lh.write(t.format(self.E.t0, " ".join(argv), printed))
|
||||||
self.logf = lh
|
self.logf = lh
|
||||||
self.logf_base_fn = base_fn
|
self.logf_base_fn = base_fn
|
||||||
print(msg, end="")
|
print(msg, end="")
|
||||||
|
|
|
@ -6,8 +6,9 @@ import re
|
||||||
import socket
|
import socket
|
||||||
import sys
|
import sys
|
||||||
|
|
||||||
from .__init__ import ANYWIN, MACOS, TYPE_CHECKING, unicode
|
from .__init__ import ANYWIN, MACOS, PY2, TYPE_CHECKING, VT100, unicode
|
||||||
from .util import chkcmd
|
from .stolen.qrcodegen import QrCode
|
||||||
|
from .util import chkcmd, sunpack
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
if TYPE_CHECKING:
|
||||||
from .svchub import SvcHub
|
from .svchub import SvcHub
|
||||||
|
@ -30,6 +31,7 @@ class TcpSrv(object):
|
||||||
self.stopping = False
|
self.stopping = False
|
||||||
self.srv: list[socket.socket] = []
|
self.srv: list[socket.socket] = []
|
||||||
self.nsrv = 0
|
self.nsrv = 0
|
||||||
|
self.qr = ""
|
||||||
ok: dict[str, list[int]] = {}
|
ok: dict[str, list[int]] = {}
|
||||||
for ip in self.args.i:
|
for ip in self.args.i:
|
||||||
ok[ip] = []
|
ok[ip] = []
|
||||||
|
@ -60,6 +62,8 @@ class TcpSrv(object):
|
||||||
for x in nonlocals:
|
for x in nonlocals:
|
||||||
eps[x] = "external"
|
eps[x] = "external"
|
||||||
|
|
||||||
|
qr1 = {}
|
||||||
|
qr2 = {}
|
||||||
msgs = []
|
msgs = []
|
||||||
title_tab: dict[str, dict[str, int]] = {}
|
title_tab: dict[str, dict[str, int]] = {}
|
||||||
title_vars = [x[1:] for x in self.args.wintitle.split(" ") if x.startswith("$")]
|
title_vars = [x[1:] for x in self.args.wintitle.split(" ") if x.startswith("$")]
|
||||||
|
@ -77,6 +81,13 @@ class TcpSrv(object):
|
||||||
|
|
||||||
msgs.append(t.format(proto, ip, port, desc))
|
msgs.append(t.format(proto, ip, port, desc))
|
||||||
|
|
||||||
|
is_ext = "external" in unicode(desc)
|
||||||
|
qrt = qr1 if is_ext else qr2
|
||||||
|
try:
|
||||||
|
qrt[ip].append(port)
|
||||||
|
except:
|
||||||
|
qrt[ip] = [port]
|
||||||
|
|
||||||
if not self.args.wintitle:
|
if not self.args.wintitle:
|
||||||
continue
|
continue
|
||||||
|
|
||||||
|
@ -86,7 +97,7 @@ class TcpSrv(object):
|
||||||
ep = "{}:{}".format(ip, port)
|
ep = "{}:{}".format(ip, port)
|
||||||
|
|
||||||
hits = []
|
hits = []
|
||||||
if "pub" in title_vars and "external" in unicode(desc):
|
if "pub" in title_vars and is_ext:
|
||||||
hits.append(("pub", ep))
|
hits.append(("pub", ep))
|
||||||
|
|
||||||
if "pub" in title_vars or "all" in title_vars:
|
if "pub" in title_vars or "all" in title_vars:
|
||||||
|
@ -110,6 +121,9 @@ class TcpSrv(object):
|
||||||
if self.args.wintitle:
|
if self.args.wintitle:
|
||||||
self._set_wintitle(title_tab)
|
self._set_wintitle(title_tab)
|
||||||
|
|
||||||
|
if self.args.qr or self.args.qrs:
|
||||||
|
self.qr = self._qr(qr1, qr2)
|
||||||
|
|
||||||
def _listen(self, ip: str, port: int) -> None:
|
def _listen(self, ip: str, port: int) -> None:
|
||||||
srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||||
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||||
|
@ -346,3 +360,41 @@ class TcpSrv(object):
|
||||||
|
|
||||||
print("\033]0;{}\033\\".format(title), file=sys.stderr, end="")
|
print("\033]0;{}\033\\".format(title), file=sys.stderr, end="")
|
||||||
sys.stderr.flush()
|
sys.stderr.flush()
|
||||||
|
|
||||||
|
def _qr(self, t1: dict[str, list[int]], t2: dict[str, list[int]]) -> str:
|
||||||
|
ip = None
|
||||||
|
for ip in list(t1) + list(t2):
|
||||||
|
if ip.startswith(self.args.qr_ip):
|
||||||
|
break
|
||||||
|
ip = ""
|
||||||
|
|
||||||
|
if not ip:
|
||||||
|
# maybe /bin/ip is missing or smth
|
||||||
|
ip = self.args.qr_ip
|
||||||
|
|
||||||
|
if not ip:
|
||||||
|
return ""
|
||||||
|
|
||||||
|
if self.args.http_only:
|
||||||
|
https = ""
|
||||||
|
elif self.args.https_only:
|
||||||
|
https = "s"
|
||||||
|
else:
|
||||||
|
https = "s" if self.args.qrs else ""
|
||||||
|
|
||||||
|
ports = t1.get(ip, t2.get(ip, []))
|
||||||
|
dport = 443 if https else 80
|
||||||
|
port = "" if dport in ports or not ports else ":{}".format(ports[0])
|
||||||
|
txt = "http{}://{}{}/{}".format(https, ip, port, self.args.qrl)
|
||||||
|
|
||||||
|
btxt = txt.encode("utf-8")
|
||||||
|
if PY2:
|
||||||
|
btxt = sunpack(b"B" * len(btxt), btxt)
|
||||||
|
|
||||||
|
qr = QrCode.encode_binary(btxt).render()
|
||||||
|
if not VT100:
|
||||||
|
return "{}\n{}".format(txt, qr)
|
||||||
|
|
||||||
|
qr = qr.replace("\n", "\033[K\n") + "\033[K" # win10do
|
||||||
|
t = "{} \033[0;38;5;{};48;5;{}m\033[J\n{}\033[999G\033[0m\033[J"
|
||||||
|
return t.format(txt, self.args.qr_fg, self.args.qr_bg, qr)
|
||||||
|
|
|
@ -16,6 +16,10 @@ https://github.com/giampaolo/pyftpdlib/
|
||||||
C: 2007 Giampaolo Rodola'
|
C: 2007 Giampaolo Rodola'
|
||||||
L: MIT
|
L: MIT
|
||||||
|
|
||||||
|
https://github.com/nayuki/QR-Code-generator
|
||||||
|
C: Project Nayuki
|
||||||
|
L: MIT
|
||||||
|
|
||||||
https://github.com/python/cpython/blob/3.10/Lib/asyncore.py
|
https://github.com/python/cpython/blob/3.10/Lib/asyncore.py
|
||||||
C: 1996 Sam Rushing
|
C: 1996 Sam Rushing
|
||||||
L: ISC
|
L: ISC
|
||||||
|
|
|
@ -24,6 +24,7 @@ copyparty/res/insecure.pem,
|
||||||
copyparty/star.py,
|
copyparty/star.py,
|
||||||
copyparty/stolen,
|
copyparty/stolen,
|
||||||
copyparty/stolen/__init__.py,
|
copyparty/stolen/__init__.py,
|
||||||
|
copyparty/stolen/qrcodegen.py,
|
||||||
copyparty/stolen/surrogateescape.py,
|
copyparty/stolen/surrogateescape.py,
|
||||||
copyparty/sutil.py,
|
copyparty/sutil.py,
|
||||||
copyparty/svchub.py,
|
copyparty/svchub.py,
|
||||||
|
|
Loading…
Reference in a new issue