Skip to main content

Malloc的深入分析与可利用点分析

· 14 min read

Consistenting fastbin before moving to smallbin. It may seem excessive to clear all fastbins before checking for available space, but it helps avoid fragmentation problems usually associated with fastbins. Additionally, in reality, programs often make consecutive small or large requests, rather than a mix of both. Thus, consolidation is not frequently needed in most programs. Programs that require frequent consolidation usually tend to fragment.

Next chunk of code will be fetching largebin.### Fragment Consolidation

    malloc_consolidate (av);

malloc_consolidate()

Defined in malloc.c at #4704

/*
------------------------- malloc_consolidate -------------------------

malloc_consolidate is a specialized version of free() that tears
down chunks held in fastbins. Free itself cannot be used for this
purpose since, among other things, it might place chunks back onto
fastbins. So, instead, we need to use a minor variant of the same
code.
*/

static void malloc_consolidate(mstate av)
{
mfastbinptr* fb; /* current fastbin being consolidated */
mfastbinptr* maxfb; /* last fastbin (for loop control) */
mchunkptr p; /* current chunk being consolidated */
mchunkptr nextp; /* next chunk to consolidate */
mchunkptr unsorted_bin; /* bin header */
mchunkptr first_unsorted; /* chunk to link to */

/* These have same use as in free() */
mchunkptr nextchunk;
INTERNAL_SIZE_T size;
INTERNAL_SIZE_T nextsize;
INTERNAL_SIZE_T prevsize;
int nextinuse;

atomic_store_relaxed (&av->have_fastchunks, false);

unsorted_bin = unsorted_chunks(av);

/*
Remove each chunk from fast bin and consolidate it, placing it
then in unsorted bin. Among other reasons for doing this,
placing in unsorted bin avoids needing to calculate actual bins
until malloc is sure that chunks aren't immediately going to be
reused anyway.
*/
/* Loop starting from the first chunk, consolidate all chunks */
maxfb = &fastbin (av, NFASTBINS - 1);
fb = &fastbin (av, 0);
do {
p = atomic_exchange_acq (fb, NULL);
if (p != 0) {
do {
{
if (__glibc_unlikely (misaligned_chunk (p))) // Pointers must be aligned
malloc_printerr ("malloc_consolidate(): "
"unaligned fastbin chunk detected");

unsigned int idx = fastbin_index (chunksize (p));
if ((&fastbin (av, idx)) != fb) // Fastbin chunk check
malloc_printerr ("malloc_consolidate(): invalid chunk size");
}

check_inuse_chunk(av, p);
nextp = REVEAL_PTR (p->fd);

/* Slightly streamlined version of consolidation code in free() */
size = chunksize (p);
nextchunk = chunk_at_offset(p, size);
nextsize = chunksize(nextchunk);

if (!prev_inuse(p)) {
prevsize = prev_size (p);
size += prevsize;
p = chunk_at_offset(p, -((long) prevsize));
/* Check if prevsize and size are equal */
if (__glibc_unlikely (chunksize(p) != prevsize))
malloc_printerr ("corrupted size vs. prev_size in fastbins");
unlink_chunk (av, p); // Unlink the prev chunk
}

if (nextchunk != av->top) {
nextinuse = inuse_bit_at_offset(nextchunk, nextsize);

if (!nextinuse) {
size += nextsize;
unlink_chunk (av, nextchunk);
} else
clear_inuse_bit_at_offset(nextchunk, 0);

/* Insert p at the head of a linked list */
first_unsorted = unsorted_bin->fd;
unsorted_bin->fd = p;
first_unsorted->bk = p;

if (!in_smallbin_range (size)) {
p->fd_nextsize = NULL;
p->bk_nextsize = NULL;
}

set_head(p, size | PREV_INUSE);
p->bk = unsorted_bin;
p->fd = first_unsorted;
set_foot(p, size);
}
// next chunk = av -> top, consolidate to topchunk
else {
size += nextsize;
set_head(p, size | PREV_INUSE);
av->top = p;
}

} while ( (p = nextp) != 0);

}
} while (fb++ != maxfb);
}

Firstly, set the PREV_INUSE of the next adjacent chunk to 1. If the previous adjacent chunk is free, merge it. Then check if the next chunk is free and merge if needed. Regardless of whether the merge is complete or not, place the fastbin or the bin after consolidation into the unsorted_bin. (If adjacent to the top chunk, merge it with the top chunk)

Iteration

So long that my head is spinning...

/*
Process recently freed or remaindered chunks, taking one only if
it is exact fit, or, if this a small request, the chunk is remainder from
the most recent non-exact fit. Place other traversed chunks in
bins. Note that this step is the only place in any routine where
chunks are placed in bins.

The outer loop here is needed because we might not realize until
near the end of malloc that we should have consolidated, so must
do so and retry. This happens at most once, and only when we would
otherwise need to expand memory to service a "small" request.
*/

#if USE_TCACHE
INTERNAL_SIZE_T tcache_nb = 0;
size_t tc_idx = csize2tidx (nb);
if (tcache && tc_idx < mp_.tcache_bins)
tcache_nb = nb;
int return_cached = 0; // Flag indicating that the appropriately sized chunk has been put into tcache

tcache_unsorted_count = 0; // Number of processed unsorted chunks

Loop through and place unsorted_bin into the corresponding bin

  for (;; )
{
int iters = 0;
while ((victim = unsorted_chunks (av)->bk) != unsorted_chunks (av)) // Have all unsorted chunks been retrieved
{
bck = victim->bk;
size = chunksize (victim);
mchunkptr next = chunk_at_offset (victim, size);
// Some safety checks
if (__glibc_unlikely (size <= CHUNK_HDR_SZ)
|| __glibc_unlikely (size > av->system_mem))
malloc_printerr ("malloc(): invalid size (unsorted)");
if (__glibc_unlikely (chunksize_nomask (next) < CHUNK_HDR_SZ)
|| __glibc_unlikely (chunksize_nomask (next) > av->system_mem))
malloc_printerr ("malloc(): invalid next size (unsorted)");
if (__glibc_unlikely ((prev_size (next) & ~(SIZE_BITS)) != size))
malloc_printerr ("malloc(): mismatching next->prev_size (unsorted)");
if (__glibc_unlikely (bck->fd != victim)
|| __glibc_unlikely (victim->fd != unsorted_chunks (av)))
malloc_printerr ("malloc(): unsorted double linked list corrupted");
if (__glibc_unlikely (prev_inuse (next)))
malloc_printerr ("malloc(): invalid next->prev_inuse (unsorted)");

/*
If a small request, try to use last remainder if it is the
only chunk in unsorted bin. This helps promote locality for
runs of consecutive small requests. This is the only
exception to best-fit, and applies only when there is
no exact fit for a small chunk.
*/


if (in_smallbin_range (nb) && // Within the small bin range
bck == unsorted_chunks (av) && // Only one chunk in unsorted_bin
victim == av->last_remainder && // Is the last remainder
(unsigned long) (size) > (unsigned long) (nb + MINSIZE)) // If size is greater than nb + MINSIZE, i.e., chunk can still be a chunk after `nb` memory is taken
{
/* split and reattach remainder */
remainder_size = size - nb;
remainder = chunk_at_offset (victim, nb); // Remaining remainder
unsorted_chunks (av)->bk = unsorted_chunks (av)->fd = remainder; // Reconstruct unsorted_bin linked list
av->last_remainder = remainder;
remainder->bk = remainder->fd = unsorted_chunks (av);
if (!in_smallbin_range (remainder_size))
{
remainder->fd_nextsize = NULL;
remainder->bk_nextsize = NULL;
}

set_head (victim, nb | PREV_INUSE |
(av != &main_arena ? NON_MAIN_ARENA : 0)); // Flag for nb
set_head (remainder, remainder_size | PREV_INUSE); // Flag for remainder
set_foot (remainder, remainder_size);

check_malloced_chunk (av, victim, nb);
void *p = chunk2mem (victim);
alloc_perturb (p, bytes);
return p; // Return nb
}

// More checks...
/* remove from unsorted list */
if (__glibc_unlikely (bck->fd != victim))
malloc_printerr ("malloc(): corrupted unsorted chunks 3");

// Retrieve the head chunk
unsorted_chunks (av)->bk = bck;
bck->fd = unsorted_chunks (av);

/* Take now instead of binning if exact fit */

if (size == nb)
{
// Set the flag
set_inuse_bit_at_offset (victim, size);
if (av != &main_arena)
set_non_main_arena (victim);
#if USE_TCACHE
/* Fill cache first, return to user only if cache fills.
We may return one of these chunks later. */
if (tcache_nb
&& tcache->counts[tc_idx] < mp_.tcache_count)
{
// Put victim into tcache instead of returning
// Since in most cases, a size that has just been needed has a higher probability of continuance, it is put into tcache
tcache_put (victim, tc_idx);
return_cached = 1;
continue;
}
else
{
#endif
check_malloced_chunk (av, victim, nb);
void *p = chunk2mem (victim);
alloc_perturb (p, bytes);
return p;
#if USE_TCACHE
}
#endif
}

/* place chunk in bin */

if (in_smallbin_range (size)) // Place into small bin
{
victim_index = smallbin_index (size);
bck = bin_at (av, victim_index);
fwd = bck->fd;
}
else
{
victim_index = largebin_index (size);
bck = bin_at (av, victim_index);
fwd = bck->fd;

/* maintain large bins in sorted order */
if (fwd != bck) // If large bin is not empty
{
/* Or with inuse bit to speed comparisons */
size |= PREV_INUSE; // Set PREV_INUSE to 1
/* if smaller than smallest, bypass loop below */
assert (chunk_main_arena (bck->bk));
/* Insert directly at the end of the large bin */
if ((unsigned long) (size)
< (unsigned long) chunksize_nomask (bck->bk))
{
fwd = bck;
bck = bck->bk;

victim->fd_nextsize = fwd->fd;
victim->bk_nextsize = fwd->fd->bk_nextsize;
fwd->fd->bk_nextsize = victim->bk_nextsize->fd_nextsize = victim;
}
else
{
assert (chunk_main_arena (fwd));
/* Find the first chunk that is not greater than `victim` */
while ((unsigned long) size < chunksize_nomask (fwd))
{
fwd = fwd->fd_nextsize;
assert (chunk_main_arena (fwd));
}

if ((unsigned long) size
== (unsigned long) chunksize_nomask (fwd))
/* If the size is the same, insert it after `fwd`, without adding nextsize to reduce computation */
/* Always insert in the second position. */
fwd = fwd->fd;
else
{
/* Insert it before `fwd`, adding nextsize */
victim->fd_nextsize = fwd;
victim->bk_nextsize = fwd->bk_nextsize;
if (__glibc_unlikely (fwd->bk_nextsize->fd_nextsize != fwd))
malloc_printerr ("malloc(): largebin double linked list corrupted (nextsize)");
fwd->bk_nextsize = victim;
victim->bk_nextsize->fd_nextsize = victim;
}
bck = fwd->bk;
if (bck->fd != fwd)
malloc_printerr ("malloc(): largebin double linked list corrupted (bk)");
}
}
else // If large bin is empty
victim->fd_nextsize = victim->bk_nextsize = victim;
}

// Insert into the linked list
mark_bin (av, victim_index);
victim->bk = bck;
victim->fd = fwd;
fwd->bk = victim;
bck->fd = victim;

#if USE_TCACHE
/* If we've processed as many chunks as we're allowed while
filling the cache, return one of the cached ones. */
// If tcache is full, get chunk from tcache
++tcache_unsorted_count;
if (return_cached
&& mp_.tcache_unsorted_limit > 0
&& tcache_unsorted_count > mp_.tcache_unsorted_limit)
{
return tcache_get (tc_idx);
}
#endif

#define MAX_ITERS 10000
if (++iters >= MAX_ITERS)
break;
}

#if USE_TCACHE
/* If all the small chunks we found ended up cached, return one now. */
// After the while loop, get chunk from tcache
if (return_cached)
{
return tcache_get (tc_idx);
}
#endif

If no chunks of the right size are found during the sorted chunk process, then find an appropriate chunk in the subsequent code

       /*
If a large request, scan through the chunks of current bin in
sorted order to find smallest that fits. Use the skip list for this.
*/

if (!in_smallbin_range (nb))
{
bin = bin_at (av, idx);

/* skip scan if empty or largest chunk is too small */
// If large bin is non-empty and the size of the first chunk >= nb
if ((victim = first (bin)) != bin
&& (unsigned long) chunksize_nomask (victim)
>= (unsigned long) (nb))
{
// Find the first chunk with size >= nb
victim = victim->bk_nextsize;
while (((unsigned long) (size = chunksize (victim)) <
(unsigned long) (nb)))
victim = victim->bk_nextsize;

/* Avoid removing the first entry for a size so that the skip
list does not have to be rerouted. */
// If `victim` is not the last one and `victim->fd` has the same size as `victim`, return next one since it is not mainained by nextsize
if (victim != last (bin)
&& chunksize_nomask (victim)
== chunksize_nomask (victim->fd))
victim = victim->fd;

remainder_size = size - nb;
unlink_chunk (av, victim); // Retrieve victim

/* Exhaust */
// If remaining is less than the minimum chunk, discard it
if (remainder_size < MINSIZE)
{
set_inuse_bit_at_offset (victim, size);
if (av != &main_arena)
set_non_main_arena (victim);
}
/* Split */
// Otherwise, split it into the unsorted_bin
else
{
remainder = chunk_at_offset (victim, nb);
/* We cannot assume the unsorted list is empty and therefore
have to perform a complete insert here. */
bck = unsorted_chunks (av);
fwd = bck->fd;
if (__glibc_unlikely (fwd->bk != bck))
malloc_printerr ("malloc(): corrupted unsorted chunks");
remainder->bk = bck;
remainder->fd = fwd;
bck->fd = remainder;
fwd->bk = remainder;
if (!in_smallbin_range (remainder_size))
{
remainder->fd_nextsize = NULL;
remainder->bk_nextsize = NULL;
}
set_head (victim, nb | PREV_INUSE |
(av != &main_arena ? NON_MAIN_ARENA : 0));
set_head (remainder, remainder_size | PREV_INUSE);
set_foot (remainder, remainder_size);
}
check_malloced_chunk (av, victim, nb);
void *p = chunk2mem (victim);
alloc_perturb (p, bytes);
return p;
}
}

/*
Search for a chunk by scanning bins, starting with next largest
bin. This search is strictly by best-fit; i.e., the smallest
(with ties going to approximately the least recently used) chunk
that fits is selected.

The bitmap avoids needing to check that most blocks are nonempty.
The particular case of skipping all bins during warm-up phases
when no chunks have been returned yet is faster than it might look.
*/
/* This part is a bit confusing */思察。t is set to 1, others are set to 0

```c
for (;; )
{
/* Skip rest of block if there are no more set bits in this block. */
/* If bit > map, it means that the free chunks in this block's bin are all smaller than the required chunk. Skip the loop directly. */
if (bit > map || bit == 0)
{
do
{
// If there are no available blocks, then use the top chunk directly
if (++block >= BINMAPSIZE) /* out of bins */
goto use_top;
}
while ((map = av->binmap[block]) == 0); // This block has no free chunks

// Find the first bin of the current block
bin = bin_at(av, (block << BINMAPSHIFT));
bit = 1;
}

/* Advance to bin with set bit. There must be one. */
// When the current bin is not available, search for the next bin
while ((bit & map) == 0)
{
bin = next_bin(bin);
bit <<= 1; // Use the next chunk
assert(bit != 0);
}

/* Inspect the bin. It is likely to be non-empty */
// Start from the smallest chunk
victim = last(bin);

/* If a false alarm (empty bin), clear the bit. */
// If the bin is empty, update the value of binmap and find the next bin
if (victim == bin)
{
av->binmap[block] = map &= ~bit; /* Write through */
bin = next_bin(bin);
bit <<= 1;
}

else
{
// If not empty, extract the chunk and perform splitting and merging
size = chunksize(victim);

/* We know the first chunk in this bin is big enough to use. */
// The first chunk (the largest one) is large enough
assert((unsigned long)(size) >= (unsigned long)(nb));

remainder_size = size - nb;

/* Unlink */
unlink_chunk(av, victim);

/* Exhaust */
if (remainder_size < MINSIZE)
{
set_inuse_bit_at_offset(victim, size);
if (av != &main_arena)
set_non_main_arena(victim);
}

/* Split */
else
{
remainder = chunk_at_offset(victim, nb);

/* We cannot assume the unsorted list is empty and therefore have to perform a complete insert here. */
bck = unsorted_chunks(av);
fwd = bck->fd;
if (__glibc_unlikely(fwd->bk != bck))
malloc_printerr("malloc(): corrupted unsorted chunks 2");
remainder->bk = bck;
remainder->fd = fwd;
bck->fd = remainder;
fwd->bk = remainder;

/* advertise as last remainder */
if (in_smallbin_range(nb))
av->last_remainder = remainder;
if (!in_smallbin_range(remainder_size))
{
remainder->fd_nextsize = NULL;
remainder->bk_nextsize = NULL;
}
set_head(victim, nb | PREV_INUSE | (av != &main_arena ? NON_MAIN_ARENA : 0));
set_head(remainder, remainder_size | PREV_INUSE);
set_foot(remainder, remainder_size);
}
check_malloced_chunk(av, victim, nb);
void *p = chunk2mem(victim);
alloc_perturb(p, bytes);
return p;
}
}
use_top:
/*
If large enough, split off the chunk bordering the end of memory
(held in av->top). Note that this is in accord with the best-fit
search rule. In effect, av->top is treated as larger (and thus
less well fitting) than any other available chunk since it can
be extended to be as large as necessary (up to system
limitations).

We require that av->top always exists (i.e., has size >=
MINSIZE) after initialization, so if it would otherwise be
exhausted by the current request, it is replenished. (The main
reason for ensuring it exists is that we may need MINSIZE space
to put in fenceposts in sysmalloc.)
*/

victim = av->top;
size = chunksize(victim);

if (__glibc_unlikely(size > av->system_mem))
malloc_printerr("malloc(): corrupted top size");
// If the top chunk can be independent after splitting nb
if ((unsigned long)(size) >= (unsigned long)(nb + MINSIZE))
{
remainder_size = size - nb;
remainder = chunk_at_offset(victim, nb);
av->top = remainder;
set_head(victim, nb | PREV_INUSE | (av != &main_arena ? NON_MAIN_ARENA : 0));
set_head(remainder, remainder_size | PREV_INUSE);

check_malloced_chunk(av, victim, nb);
void *p = chunk2mem(victim);
alloc_perturb(p, bytes);
return p;
}

/* When we are using atomic ops to free fast chunks we can get
here for all block sizes. */
// If it is not enough to split and there are still fastbins, merge the fastbins
else if (atomic_load_relaxed(&av->have_fastchunks))
{
malloc_consolidate(av);
/* Restore the original bin index */
if (in_smallbin_range(nb))
idx = smallbin_index(nb);
else
idx = largebin_index(nb);
}

/*
Otherwise, relay to handle system-dependent cases
*/
// Otherwise, call sysmalloc to request memory from the operating system
else
{
void *p = sysmalloc(nb, av);
if (p != NULL)
alloc_perturb(p, bytes);
return p;
}
}
info

This Content is generated by ChatGPT and might be wrong / incomplete, refer to Chinese version if you find something wrong.

MiBand-8-Pro-Data-to-Obsidian

· 10 min read

Recently, I set up a life management system with the help of DIYGOD. With various plugins, I achieved semi-automation. However, manually recording sleep time, steps, and other data like heart rate and blood pressure is not very geeky. After some research, I found out that Zepp (formerly Huami) has a reverse-engineered API interface that stores step count and other information in plaintext. This led me to impulsively purchase the Xiaomi Mi Band 8 Pro Genshin Impact Limited Edition. To my surprise, I discovered that the Xiaomi Mi Band 8 no longer supports Zepp. Although the Xiaomi Mi Band 7 does not officially support Zepp, it can still be used by modifying the QR code and using the Zepp installation package. However, the Xiaomi Mi Band 8 has completely deprecated Zepp.

Initial Exploration — Packet Capture

Firstly, I attempted to capture packets to see if there was any useful information available. I used to use Proxifier for packet capture, but it was not very effective due to some software having SSLPinning. This time, I utilized mitmproxy along with a system-level certificate.

Tools Used

Testing Method

In a nutshell, I installed mitmproxy on my PC, obtained the mitmproxy-ca-cert.cer file in the $HOME/.mitmproxy directory, and installed it on the Android device as per the normal workflow.

I then installed ConscryptTrustUserCerts in Magisk, restarted the device, which mounted the user-level certificate to the system-level certificate directory during boot. This completed the preparation.

After opening mitmweb on the PC, setting the Wi-Fi proxy on the phone to <my-pc-ip>:8080, I successfully captured HTTPS requests.

Conclusion

It was not very useful. All requests were encrypted, and there were signatures, hashes, nonces, etc., to ensure security. I did not want to reverse engineer the apk, so I abandoned this approach.

Glimpse of Hope — BLE Connection

Since packet capturing was not feasible, I decided to create a BLE client to connect to the smart band and retrieve data, which seemed like a very reasonable approach. Moreover, this method did not require any actions on my phone; a script running on Obsidian, with one connection and data retrieval, seemed to be very automated.

Implementation

The code mainly referenced wuhan005/mebeats: 💓 Real-time heart rate data collection for Xiaomi Mi Bands. However, as his tools were for MacOS, I made some modifications with the help of GPT.

// Java code block translated to English
public final void bindDeviceToServer(lg1 lg1Var) {

Logger.i(getTAG(), "bindDeviceToServer start");

HuaMiInternalApiCaller huaMiDevice = HuaMiDeviceTool.Companion.getInstance().getHuaMiDevice(this.mac);

if (huaMiDevice == null) {

String tag = getTAG();

Logger.i(tag + "bindDeviceToServer huaMiDevice == null", new Object[0]);

if (lg1Var != null) {

lg1Var.onConnectFailure(4);

}

} else if (needCheckLockRegion() && isParallel(huaMiDevice)) {

unbindHuaMiDevice(huaMiDevice, lg1Var);

} else {

DeviceInfoExt deviceInfo = huaMiDevice.getDeviceInfo();

if (deviceInfo == null) {

String tag2 = getTAG();

Logger.i(tag2 + "bindDeviceToServer deviceInfo == null", new Object[0]);

return;

}

String sn = deviceInfo.getSn();

setMDid("huami." + sn);

setSn(deviceInfo.getSn());

BindRequestData create = BindRequestData.Companion.create(deviceInfo.getSn(), this.mac, deviceInfo.getDeviceId(), deviceInfo.getDeviceType(), deviceInfo.getDeviceSource(), deviceInfo.getAuthKey(), deviceInfo.getFirmwareVersion(), deviceInfo.getSoftwareVersion(), deviceInfo.getSystemVersion(), deviceInfo.getSystemModel(), deviceInfo.getHardwareVersion());

String tag3 = getTAG();

Logger.d(tag3 + create, new Object[0]);

getMHuaMiRequest().bindDevice(create, new HuaMiDeviceBinder$bindDeviceToServer$1(this, lg1Var), new HuaMiDeviceBinder$bindDeviceToServer$2(lg1Var, this));

}

}

By examining this function, we can see that the data is retrieved from deviceInfo, which is obtained from huaMiDevice. For those interested, the details of how this is derived can be explored in the package com.xiaomi.wearable.wear.connection.

The Ultimate Solution — Frida Hook

At this point, I had already decided on the final approach - reverse engineering. Since the data sent out is encrypted, there must be a process where unencrypted data handling occurs. By reverse engineering it, hooking into it, and writing an Xposed module to monitor it, the task could be accomplished.

Due to time constraints, I will not delve into how to install Frida.

Initially, I used jadx-gui with the feature copy as frida snippets, which saved a lot of effort. However, due to various peculiarities of Kotlin data classes, many times the necessary information cannot be obtained. As I did not document my journey while troubleshooting, here is a brief overview:

  1. Initially, I observed the fitness_summary database in the /data/data/com.mi.health/databases folder, which contains the desired data. Cross-referencing led me to the com.xiaomi.fit.fitness.persist.db.internal class.
  2. Exploring methods such as update and insert, I found com.xiaomi.fit.fitness.persist.db.internal.h.getDailyRecord method which had output every time a refresh occurred, but only contained values such as sid, time, and did not include the value.
  3. Continuing the trail, I used the given code snippet to inspect overloads and parameter types.
var insertMethodOverloads = hClass.updateAll.overloads;

for (var i = 0; i < insertMethodOverloads.length; i++) {
var overload = insertMethodOverloads[i];
console.log("Overload #" + i + " has " + overload.argumentTypes.length + " arguments.");
for (var j = 0; j < overload.argumentTypes.length; j++) {
console.log(" - Argument " + j + ": " + overload.argumentTypes[j].className);
}
}
  1. It struck me that exceptions could be utilized to examine the function call stack - a breakthrough moment.
var callerMethodName = Java.use("android.util.Log").getStackTraceString(Java.use("java.lang.Exception").$new());
console.log("getTheOneDailyRecord called by: " + callerMethodName);
  1. Proceeding layer by layer, I discovered the class com.xiaomi.fit.fitness.export.data.aggregation.DailyBasicReport, which perfectly met my needs.
    dbutilsClass.getAllDailyRecord.overload('com.xiaomi.fit.fitness.export.data.annotation.HomeDataType', 'java.lang.String', 'long', 'long', 'int').implementation = function (homeDataType, str, j, j2, i) {
console.log("getAllDailyRecord called with args: " + homeDataType + ", " + str + ", " + j + ", " + j2 + ", " + i);
var result = this.getAllDailyRecord(homeDataType, str, j, j2, i);
var entrySet = result.entrySet();
var iterator = entrySet.iterator();
while (iterator.hasNext()) {
var entry = iterator.next();
console.log("entry: " + entry);
}
var callerMethodName = Java.use("android.util.Log").getStackTraceString(Java.use("java.lang.Exception").$new());
console.log("getTheOneDailyRecord called by: " + callerMethodName);
return result;
}

// Output: DailyStepReport(time=1706745600, time = 2024-02-01 08:00:00, tag='days', steps=110, distance=66, calories=3, minStartTime=1706809500, maxEndTime=1706809560, avgStep=110, avgDis=66, active=[], stepRecords=[StepRecord{time = 2024-02-02 01:30:00, steps = 110, distance = 66, calories = 3}])
  1. Faced a challenge as steps is a private attribute, and none of the interfaces like getSteps(), getSourceData() worked, all displaying not a function. Likely a difference in Kotlin and Java handling. Resorted to using reflection for resolution.

The final frida script was formulated to fetch the daily steps data. Altering HomeDataType would yield other data.

var CommonSummaryUpdaterCompanion = Java.use("com.xiaomi.fitness.aggregation.health.updater.CommonSummaryUpdater$Companion");
var HomeDataType = Java.use("com.xiaomi.fit.fitness.export.data.annotation.HomeDataType");
var instance = CommonSummaryUpdaterCompanion.$new().getInstance();
console.log("instance: " + instance);

var step = HomeDataType.STEP;
var DailyStepReport = Java.use("com.xiaomi.fit.fitness.export.data.aggregation.DailyStepReport");

var result = instance.getReportList(step.value, 1706745600, 1706832000);
var report = result.get(0);
console.log("report: " + report + report.getClass());


var stepsField = DailyStepReport.class.getDeclaredField("steps");
stepsField.setAccessible(true);
var steps = stepsField.get(report);
console.log("Steps: " + steps);
// Output: Steps: 110

Final – Xposed Module

The approach now is to listen to a specific address using XPosed, and then to slightly protect against plaintext transmission pigeonholed here. Since the app is always active, I believe this method is feasible. The current challenge is my lack of knowledge in writing Kotlin, let alone Xposed.

Fortunately, the Kotlin compiler's suggestions are powerful enough, and besides configuring Xposed, no additional knowledge is required. Coupled with the powerful GPT, I spent an hour or two figuring out the initial environment setup (hard to assess gradle, it's slow without a proxy, and with a proxy, it becomes unresponsive).```kotlin if (record != null) { SerializableStepRecord( time = XposedHelpers.getLongField(record, "time"), steps = XposedHelpers.getIntField(record, "steps"), distance = XposedHelpers.getIntField(record, "distance"), calories = XposedHelpers.getIntField(record, "calories") ) } else null }

    val activeStageList = activeStageListObject.mapNotNull { activeStageItem ->
if (activeStageItem != null) {
SerializableActiveStageItem(
calories = XposedHelpers.getIntField(activeStageItem, "calories"),
distance = XposedHelpers.getIntField(activeStageItem, "distance"),
endTime = XposedHelpers.getLongField(activeStageItem, "endTime"),
riseHeight = XposedHelpers.getObjectField(activeStageItem, "riseHeight") as? Float,
startTime = XposedHelpers.getLongField(activeStageItem, "startTime"),
steps = XposedHelpers.getObjectField(activeStageItem, "steps") as? Int,
type = XposedHelpers.getIntField(activeStageItem, "type")
)
} else null
}

return SerializableDailyStepReport(
time = XposedHelpers.getLongField(xposedReport, "time"),
tag = XposedHelpers.getObjectField(xposedReport, "tag") as String,
steps = XposedHelpers.getIntField(xposedReport, "steps"),
distance = XposedHelpers.getIntField(xposedReport, "distance"),
calories = XposedHelpers.getIntField(xposedReport, "calories"),
minStartTime = XposedHelpers.getObjectField(xposedReport, "minStartTime") as Long?,
maxEndTime = XposedHelpers.getObjectField(xposedReport, "maxEndTime") as Long?,
avgStep = XposedHelpers.callMethod(xposedReport, "getAvgStepsPerDay") as Int,
avgDis = XposedHelpers.callMethod(xposedReport, "getAvgDistancePerDay") as Int,
stepRecords = stepRecords,
activeStageList = activeStageList
)
}

}


The code above shows a function that processes data retrieved from some records and returns a `SerializableDailyStepReport` object. It extracts and maps various attributes from the records, such as time, steps, distance, and calories, into corresponding fields of the `SerializableStepRecord` and `SerializableActiveStageItem` objects. Finally, it constructs a `SerializableDailyStepReport` object with the processed data.

```kotlin
// build.gradle.kts [Module]
plugins {
...
kotlin("plugin.serialization") version "1.9.21"
}

dependencies {
...
implementation("org.jetbrains.kotlinx:kotlinx-serialization-json:1.6.2")
}

The first code snippet contains the configuration in the build.gradle.kts file for enabling the Kotlin serialization plugin. It also includes the dependency for kotlinx-serialization-json library for JSON serialization.

return Json.encodeToJsonElement(SerializableDailyStepReport.serializer(), convertToSerializableReport(today))

In the above statement, it uses Json.encodeToJsonElement to convert a SerializableDailyStepReport object to a JSON element using its serializer.

Broadcasting

The discussion in this section delves into the challenges faced while considering broadcasting data for an Android application. The initial idea was to use a BroadcastReceiver but was dropped due to complexities related to sending messages between the Android device and a computer.

This led to exploring alternatives like HTTP RESTful APIs, which were implemented using Ktor. However, the fluctuating data retrieval schedule and the need for continuous server upkeep introduced concerns regarding power consumption.

Subsequently, the notion of using sockets was explored to establish communication. A ServerSocket is created to listen for incoming connections, and a ClientHandler is spawned to handle each client's requests. This approach provides a more direct and energy-efficient means of communication compared to HTTP servers.

class MySocketServer(
private val port: Int,
private val lpparam: LoadPackageParam,
private val instance: Any
) {
fun startServerInBackground() {
Thread {
try {
val serverSocket = ServerSocket(port)
Log.d("MiBand", "Server started on port: ${serverSocket.localPort}")
while (!Thread.currentThread().isInterrupted) {
val clientSocket = serverSocket.accept()
val clientHandler = ClientHandler(clientSocket)
Thread(clientHandler).start()
}
} catch (e: Exception) {
Log.e("MiBand", "Server Error: ${e.message}")
}
}.start()
}

Above is a snippet depicting the creation of a socket server that listens on a specified port, handles incoming client connections, and delegates processing to separate threads for improved concurrency.

The subsequent realization of the limitation concerning running external scripts in the Obsidian environment using Templater led to the manual implementation of HTTP protocol communication to cater to data retrieval requirements within that context.

override fun run() {
try {
// Code for handling HTTP requests and responses
} catch (e: IOException) {
e.printStackTrace()
}
}

private fun parseQueryString(query: String?): Map<String, String> {
// Parsing the query string from the HTTP request
}

private fun sendSuccessResponse(outputStream: PrintWriter, result: SerializableResponse) {
// Sending a successful HTTP response with serialized data
}

The code snippet above demonstrates the processing of incoming HTTP requests by parsing the request, handling different paths, and sending appropriate responses back to the clients.

Overall, the combined use of socket communication and manual HTTP handling provides the necessary infrastructure to facilitate data exchange between the Android application and external systems while maintaining a balance between efficiency and functionality.

info

This Content is generated by ChatGPT and might be wrong / incomplete, refer to Chinese version if you find something wrong.

Wayland---腾讯会议屏幕共享解决方案

· 4 min read

Wayland - Tencent Meeting Screen Sharing Solution

During a team meeting, I tried to share my screen but only my mouse pointer was visible. In the end, it turned into using a robust phone camera solution, which was not ideal. After some searching, I found a relatively elegant (albeit twisted) solution, so I decided to document it briefly.

PWN Debugging and 1-day exploit development for CVE-2018-1160

· 5 min read
MuelNova
Pwner who wants to write codes.

Attachment download link: https://pwnable.tw/static/chall/netatalk.tgz + https://pwnable.tw/static/libc/libc-18292bd12d37bfaf58e8dded9db7f1f5da1192cb.so

It took about 1.5 days, and overall it was a very productive debugging and reproducing process. I learned some exploitation and debugging techniques, and it was very helpful for expanding my mindset.

The discovery process of the vulnerability is explained clearly by the author in Exploiting an 18 Year Old Bug. A Write-up for CVE-2018–1160 | by Jacob Baines, which is very interesting. You can also find a translated version at Discovery and Exploitation of Netatalk CVE-2018-1160_c01dkit's Blog-CSDN Blog.

The author mentioned in their blog that this vulnerability can only be exploited on NAS with -no-pie. However, the creator of the HITCON 2019 challenge, DDAA, provided an exploit approach in HITCON CTF 2019 Pwn 371 Netatalk (ddaa.tw), which basically involves leveraging the nature of fork where child processes do not change the memory layout — in other words, ASLR plays a very minor role (laughs). This way, we can expose a valid address through a side channel and then exploit it.

PWN CVE-2023-4911 Reproduction

· 9 min read
MuelNova
Pwner who wants to write codes.

Recently encountered this vulnerability, it seems to have a wide range of potential exploits. Although most machines in China seem to have a relatively low version of libc, let's take a look at it first.

Environment Setup

Testing Environment

OS: Ubuntu 22.04.1 LTS on Windows 10 x86_64

Kernel: 5.15.123.1-microsoft-standard-WSL2

Glibc: 2.35-0ubuntu3.3

从零开始的-Python-AsyncIO-生活

· 12 min read

From Zero to Python AsyncIO Life

I've been using AsyncIO for asynchronous programming in Python, but I've never thought about why. Let's take this opportunity to understand AsyncIO better.

Iterable

First of all, we need to understand what an Iterable is, which is basically an object that can be used in a for loop. Common examples of Iterable include list, str, tuple, and dict.

In Python, how does it determine if an object is an Iterable? We can use the dir() function to check its attribute list.

By running the following code, we can see their common interface:

from typing import Iterable

iterable = [
"", # str
[], # list
{}, # dict
(), # tuple
set() # set
]

def show_diff(*objects: Iterable):
"""Print the attribute differences between Iterable and object"""
assert objects
attrs = set(dir(objects[0]))
for obj in objects[1:]:
attrs &= set(dir(obj)) # Get the intersection of Iterables
attrs -= set(dir(object)) # Get the difference between Iterable and object
print(attrs)

show_diff(*iterable)

# {'__iter__', '__contains__', '__len__'}

As we can see, the key attribute is __iter__. In fact, for any object that has the __iter__ method specified, it will be considered an Iterable. Attributes like __len__ and __contains__ are common to container type Iterables.

If we add a non-container type Iterable, the result becomes obvious:

iterable = [
"", # str
[], # list
{}, # dict
(), # tuple
set(), # set
open(__file__) # IO
]

show_diff(*iterable)

# {'__iter__'}

Iterator

In Python, methods like __iter__ in Iterables have corresponding calling methods, which is iter().

Let's see the results when we use iter() on the container type Iterables listed above:

for i in iterable:
print(iter(i))

"""
<str_iterator object at 0x7f7bd06fafe0>
<list_iterator object at 0x7f7bd06fafe0>
<dict_keyiterator object at 0x7f7bd08c4b80>
<tuple_iterator object at 0x7f7bd06fafe0>
<set_iterator object at 0x7f7bd0720440>
"""

We can see that they all return an Iterator object. As demonstrated in the Iterable section, let's once again find the attribute differences among them:

# {'__next__', '__iter__'}

So, compared to Iterable, there is an additional __next__ method in Iterator, which is used to return data in the next iteration.

In the end, after all values have been iterated, it will raise a StopIteration error to indicate the end of the iteration.

We can build a custom Iterator with the following code:

class MyIterator:
def __init__(self, Iter):
self.index = 0
self.data = Iter

def __next__(self):
while self.index < len(self.data):
data = self.data[self.index]
self.index += 1
return data
raise StopIteration

def __iter__(self):
"""Iterators must be iterable"""
return self

things = ["I", "AM", "ITERABLE", "GOD"]

for i in MyIterator(things):
print(i)

Stay tuned for the next parts!```python task...") t1 = time.time() await Awaitable(sleep, 2) assert time.time() - t1 > 2, "You didn't block, silly pig" print(" I'm finished") return 123

class Awaitable: def init(self, *obj): self.obj = obj

def __await__(self):
yield self.obj

class Task: def init(self, _task): self.coro = _task

def run(self):
while True:
try:
x = self.coro.send(None)
except StopIteration as _e:
result = _e.value
break
else:
func, arg = x
func(arg)
return result

Task(task()).run()


Returning to our `small_step`, we are using a hard-coded blocking mechanism `sleep(2)`, but in reality, there are more types of blocking than just this one. We should aim for a more general mechanism for blocking.

In `Awaitable`, we are directly yielding `self`.

```python
class Awaitable:
def __init__(self, *obj):
self.obj = obj

def __await__(self):
yield self

class Task:
def __init__(self, _task):
self.coro = _task

def run(self):
while True:
try:
x = self.coro.send(None)
except StopIteration as _e:
result = _e.value
break
else:
func, arg = x.obj
func(arg)
return result

Now, notice one thing: our Task.run() function is still blocking, and we haven't completely yielded control of our program's execution. Let's continue to modify the Task code.

class Task:
def __init__(self, _task):
self.coro = _task
self._done = False
self._result = None

def run(self):
if not self._done:
try:
x = self.coro.send(None)
except StopIteration as _e:
self._result = _e.value
self._done = True
else:
... # This should not happen, an exception should be raised


t = Task(task())
t.run()
for i in range(10): # During sleep(2), we can do other things.
print("doing something", i)
sleep(0.2)
t.run()

We are manually scheduling multiple tasks here. In reality, we should schedule tasks automatically through an event loop (Event Loop).

Event Loop

Firstly, tasks must have a queue. We can use a deque double-ended queue to store tasks.

class Event:
def __init__(self):
self._queue = collections.deque()

def call_soon(self, callback, *args, **kwargs):
self._queue.append((callback, args, kwargs))

Next, we add scheduled tasks. Due to the special nature of scheduled tasks, we use a heap to store them. Here, we leverage heapq for operations.

class Event:
def __init__(self):
self._queue = collections.deque()
self._scheduled = []

def call_soon(self, callback, *args, **kwargs):
self._queue.append((callback, args, kwargs))

def call_later(self, delay, callback, *args, **kwargs):
_t = time.time() + delay
heapq.heappush(self._scheduled, (_t, callback, args, kwargs))

Let's write the event scheduling function.

class Event:
def __init__(self):
self._queue = collections.deque()
self._scheduled = []
self._stopping = False

def call_soon(self, callback, *args, **kwargs):
self._queue.append((callback, args, kwargs))

def call_later(self, delay, callback, *args, **kwargs):
_t = time.time() + delay
heapq.heappush(self._scheduled, (_t, callback, args, kwargs))

def stop(self):
self._stopping = True

def run_forever(self):
while True:
self.run_once() # At least one execution is necessary, so put the condition check below
if self._stopping:
break

def run_once(self):
now = time.time()
if self._scheduled and now > self._scheduled[0][0]:
_, cb, args, kwargs = heapq.heappop(self._scheduled)
self._queue.append((cb, args, kwargs))

task_num = len(self._queue) # Prevent adding more tasks to the queue during execution
for _ in range(task_num):
cb, args, kwargs = self._queue.popleft()
cb(*args, **kwargs)


t = Task(task())
loop = Event()
loop.call_soon(t.run)
loop.call_later(2, t.run)
loop.call_later(2.1, loop.stop)
loop.run_forever()

Now, let's modify small_step

async def small_step():
t1 = time.time()
time_ = random.randint(1, 3)
await Awaitable(time_)
assert time.time() - t1 > time_, f"{time_} You didn't block, silly pig {time.time() - t1}"
return time_

As this time has been passed to Task, we need to handle it in Task, which means adding a loop.call_later() while blocking.

class Task:
def __init__(self, _task):
self.coro = _task
self._done = False
self._result = None

def run(self):
if not self._done:
try:
x = self.coro.send(None)
except StopIteration as _e:
self._result = _e.value
self._done = True
else:
loop.call_later(*x.obj, self.run)
else:
... # This should not happen, an exception should be raised

Now, we can remove the manually specified call_later

t = Task(task())
loop = Event()
loop.call_soon(t.run)
loop.call_later(1.1, loop.stop) # random() will only yield values between 0 and 1
loop.run_forever()

Finally, let's try implementing multiple tasks and actually demonstrate the async effect through some parameters.

import collections
import heapq
import itertools
import random
import time
from time import sleep

count = itertools.count(0)
total = 0


async def task():
""" Create a new task """
print("TASK BEGIN...")

print(" MainStep...")

main_result = await main_step()

print(f" MainStep Finished with result {main_result}")

print("TASK END")


async def main_step():
print(" SmallStep(s)...")

small_result = await small_step()

print(f" SmallStep(s) Finished with result {small_result}")

return small_result * 100


async def small_step():
t1 = time.time()
time_ = random.random()
await Awaitable(time_)
assert time.time() - t1 > time_, f"{time_} You didn't block, silly pig {time.time() - t1}"
return time_


class Awaitable:
def __init__(self, *obj):
self.obj = obj

def __await__(self):
yield self


class Task:
def __init__(self, _task):
self.coro = _task
self._done = False
self._result = None
self._id = f"Task-{next(count)}"

def run(self):
print(f"--------- {self._id} --------")
if not self._done:
try:
x = self.coro.send(None)
except StopIteration as _e:
self._result = _e.value
self._done = True
else:
loop.call_later(*x.obj, self.run)
else:
... # This should not happen, an exception should be raised
print("-------------------------")


class Event:
def __init__(self):
self._queue = collections.deque()
self._scheduled = []
self._stopping = False

def call_soon(self, callback, *args, **kwargs):
self._queue.append((callback, args, kwargs))

def call_later(self, delay, callback, *args, **kwargs):
_t = time.time() + delay
global total
total += delay
heapq.heappush(self._scheduled, (_t, callback, args, kwargs))

def stop(self):
self._stopping = True

def run_forever(self):
while True:
self.run_once() # At least one execution is necessary, so put the condition check below
if self._stopping:
break

def run_once(self):
now = time.time()
if self._scheduled and now > self._scheduled[0][0]:
_, cb, args, kwargs = heapq.heappop(self._scheduled)
self._queue.append((cb, args, kwargs))

task_num = len(self._queue) # Prevent adding more tasks to the queue during execution
for _ in range(task_num):
cb, args, kwargs = self._queue.popleft()
cb(*args, **kwargs)


t = Task(task())
loop = Event()
loop.call_soon(t.run)
loop.call_later(1.1, loop.stop)
loop.run_forever()

Here, we can see that while we would normally need around 509.3s to run all the tasks, thanks to the concurrent execution achieved through task scheduling, we finished running all 1000 tasks within just 1 second.

Future

Finally, our code actively uses sleep to simulate blocking. How should we do this in a real-world scenario?

Typically, we want to perform an operation and obtain a value, as shown below:

async def small_step():
result = await Awaitable(...)
return result

In this situation, we should introduce Future. What is a Future? It's a result that will happen in the future, as opposed to Awaitable, where we can't pass the result at the time of creation.

class Future:
def __init__(self):
self._result = None
self._done = False

def set_result(self, result):
if self._done:
raise RuntimeError() # Disallowed operation
self._result = result
self._done = True

@property
def result(self):
if self._done:
return self._result
raise RuntimeError()

def __await__(self):
yield self

Therefore, we need something to designate when to execute set_result.

async def small_step():
fut = Future()
# do something that will call set_result
...
result = await fut
return result

In this case, Task should receive this future, but the future doesn't have any information, only a flag telling us the task is not yet completed.

How does our Task know when to resume execution?

We can add a callback record in Future to signify this.

class Future:
def __init__(self):
self._result = None
self._done = False
self._callbacks = []

def add_done_callback(self, cb):
self._callbacks.append(cb)

def set_result(self, result):
if self._done:
raise RuntimeError() # Disallowed operation
self._result = result
self._done = True

for cb in self._callbacks:
cb() # May have other parameters

@property
def result(self):
if self._done:
return self._result
raise RuntimeError()

def __await__(self):
yield self
return self.result # result = await fut will retrieve this value

class Task:
def __init__(self, _task):
self.coro = _task
self._done = False
self._result = None
self._id = f"Task-{next(count)}"

def run(self):
print(f"--------- {self._id} --------")
if not self._done:
try:
x = self.coro.send(None)
except StopIteration as _e:
self._result = _e.value
self._done = True
else:
x.add_done_callback(self.run)
else:
... # This should not happen, an exception should be raised
print("-------------------------")

Now, we can observe Task and Future

We can see that Task can simply inherit from Future.

class Task(Future):
def __init__(self, _task):
super().__init__()
self.coro = _task
self._id = f"Task-{next(count)}"

def run(self):
print(f"--------- {self._id} --------")
if not self._done:
try:
x = self.coro.send(None)
except StopIteration as _e:
self.set_result(_e.value)
else:
x.add_done_callback(self.run)
else:
... # This should not happen, an exception should be raised
print("-------------------------")

At this point, AsyncIO is basically implemented. However, compared to Python's own AsyncIO, our code could be considered very basic. It lacks in performance (since it's not written in C) and has issues in exception handling and other areas. Finally, here is the optimized code. (Didn't mention the hook-up between Task and loop, but it's written)

import collections
import heapq
import itertools
import random
import threading
import time
from time import sleep

count = itertools.count(0)
blocked = 0


async def task():
""" Create a new task """
print("TASK BEGIN...")

print(" MainStep...")

main_result = await main_step()

print(f" MainStep Finished with result {main_result}")

print("TASK END")


async def main_step():
print(" SmallStep(s)...")

small_result = await small_step()

print(f" SmallStep(s) Finished with result {small_result}")

return small_result * 100


async def small_step():
fut = Future()
fake_io(fut)
result = await fut
return result


class Future:
def __init__(self):
self._result = None
self._done = False
self._callbacks = []

def add_done_callback(self, cb):
self._callbacks.append(cb)

def set_result(self, result):
if self._done:
raise RuntimeError() # Disallowed operation
self._result = result
self._done = True

for cb in self._callbacks:
cb() # May have other parameters

@property
def result(self):
if self._done:
return self._result
raise RuntimeError()

def __await__(self):
yield self
return self.result


class Task(Future):
def __init__(self, _task):
super().__init__()
self._loop = loop
self.coro = _task
self._id = f"Task-{next(count)}"
self._loop.call_soon(self.run)
self._start_time = time.time()

def run(self):
print(f"--------- {self._id} --------")
if not self._done:
try:
x = self.coro.send(None)
except StopIteration as _e:
self.set_result(_e.value)
global blocked
blocked += time.time() - self._start_time
else:
x.add_done_callback(self.run)
else:
... # This should not happen, an exception should be raised
print("-------------------------")


class Event:
def __init__(self):
self._queue = collections.deque()
self._scheduled = []
self._stopping = False

def call_soon(self, callback, *args, **kwargs):
self._queue.append((callback, args, kwargs))

def call_later(self, delay, callback, *args, **kwargs):
_t = time.time() + delay
heapq.heappush(self._scheduled, (_t, callback, args, kwargs))

def stop(self):
self._stopping = True

def run_forever(self):
while True:
self.run_once() # At least one execution is necessary, so put the condition check below
if self._stopping:
break

def run_once(self):
now = time.time()
if self._scheduled and now > self._scheduled[0][0] + (10 ** -5):
_, cb, args, kwargs = heapq.heappop(self._scheduled)
self._queue.append((cb, args, kwargs))

task_num = len(self._queue) # Prevent adding more tasks to the queue during execution
for _ in range(task_num):
cb, args, kwargs = self._queue.popleft()
cb(*args, **kwargs)


def fake_io(fut):
def read():
sleep(t_ := random.random()) # IO blocking
fut.set_result(t_)
threading.Thread(target=read).start()


def run_until_all_task(tasks):
if tasks := [_task for _task in tasks if not _task._done]:
loop.call_soon(run_until_all_task, tasks)
else:
loop.call_soon(loop.stop)


loop = Event()
all_tasks = [Task(task()) for _ in range(1000)]
loop.call_soon(run_until_all_task, all_tasks)
t1 = time.time()
loop.run_forever()
print(time.time() - t1, blocked)

info

This Content is generated by ChatGPT and might be wrong / incomplete, refer to Chinese version if you find something wrong.

Automatic Blog Deployment Using Github Webhook

· 2 min read
MuelNova
Pwner who wants to write codes.

Feeling tired of typing ssh, git pull, and npm run build every time to deploy your blog? Well, I thought about using a webhook.

But still need to manually resolve the conflicts in package-lock.json, wondering if there's a way to ignore it (just ignore this thing!)

Create WSL2 Bridged Network with Hyper-V and IPv6 Support

· 2 min read
MuelNova
Pwner who wants to write codes.

I don't know what happened, I used to be able to access services in WSL2 directly from the host using localhost:port, but suddenly it doesn't work today. Taking this opportunity, I'll create a virtual network card based on the documentation I read recently to set up a WSL2 bridge, which will not only support IPv6 but also allow direct access to my WSL2 services in the local network without the need for port forwarding. Since my WSL2 is not sandboxed anyway, I'm not too concerned about security xD.

Installing Arch Linux on USB External Hard Drive and Reserving Storage Space for Windows Devices

· 5 min read
MuelNova
Pwner who wants to write codes.

Frustrated with syncing data across multiple platforms for years, I suddenly remembered that I still have a SanDisk 256G Gen3.1 USB flash drive I can use, so I decided to research how to install Arch Linux on a USB drive.

Preparation:

  • VMWare Workstation
  • Arch Linux image
  • A fast and large capacity USB drive (recommended USB 3.0+, with a size of 50GB or more)