Linux에서 Crash 제어 (feat, setsetjmp/siglongjmp)

* DroidKnights 2021 에서 발표했던 내용 일부를 발췌한 글 입니다.*

Native 코드를 개발할 때,

특히 Android 와 같이 개발사/기기사 간 특수성이 존재하는 개발환경에서는 시스템에 밀접한 작업을 하는 보안 모듈 특성 때문에 충분한 테스트가 늘 아쉽기 마련이다. 혹시 모를 경우를 대비해 Crash를 Handling함으로서 적어도 Crash 때문에 배포날의 긴장도를 낮춰보도록 하자.

 

예제 코드 : https://github.com/ParkHanbum/mystudy/tree/master/signal_handler

 

 

시나리오는 다음과 같다.

 

1. 갱신된 함수를 실행하기 전에 signal 들에 대한 handling  추가.

2. signal 이 발생했을 경우를 위한 복원점 지정

3. signal 이 발생했을 경우 복원점으로 복원

4. signal 이 발생했을 경우 기존 함수 호출

 

__attribute__((optimize("-O1")))
void do_segv()
{
    fprintf(stderr, "%s\n", __func__);
    int *segv = 0;
    *segv = 1;

    MARK();
}

__attribute__((optimize("-O1")))
void do_segv1()
{
    fprintf(stderr, "%s\n", __func__);
    int *segv = 0;
    *segv = 1;

    MARK();
}

__attribute__((optimize(0)))
void do_signal_handling()
{
    int sig = 0;
    int watch_signals = sigill | sigtrap | sigabrt | sigbus | sigsegv;
    fprintf(stderr, "check : %lx\n", get_func_size((unsigned long)&do_segv));
    TRY(sig, watch_signals, do_segv1) {
        //sleep(10);
        do_segv1();
    }
    CATCH (sig, sigsegv, e) {
        fprintf(stderr, "inside catch!!\n");
        fprintf(stderr, "catch!! signal %d\n", sig);
        fprintf(stderr, "check %d  %lx \n", e.code, e.addr);
    }
    FINAL(watch_signals);
}

int main() {
    fprintf(stderr, "[%d] INIT NATIVE!!!!!!!!!!!!!!!!\n", getpid());
    unittest();

    int pid = fork();
    if (pid) {
        do_signal_handling();
        do_segv();
    } else {
        sleep(1);
    }
    return 0;
}

do_segv, do_segv1 모두 segfault를 발생시킬 함수인데, 차이점은 TRY-CATCH 매크로로 signal을 캐치하고 sigsetjmp 를 통해 설정해놓은 지점으로 siglongjmp 로 복귀하는 것.

 

/**
 * int sigsetjmp(sigjmp_buf env, int savesigs);
 * If, and only if, the savesigs argument provided to sigsetjmp() is nonzero,
 * the process's current signal mask is saved in env and will be restored
 * if a siglongjmp() is later performed with this env.
 */
#define TRY(sig, order, func)        \
    set_signal_handler(order); \
    set_watch_func((unsigned long)&func, get_func_size((unsigned long)&func));       \
    sig = sigsetjmp(point, 1); \
    if (sig == 0)

#define CATCH(sig, sigorder, exception) \
    sig_order handle_sigorder = sigorder; \
    if (sig_to_order(sig) && sigorder)

#define FINAL(order) \
    unset_signal_handler(order);

TRY-CATCH-FINAL의역할은다음과같다.

TRY

- 처리할 signal을 지정

- signal이 지정된 함수의 range를 계산

- signal이 발생했을때 지정된 함수인지 확인, sigsetjmp로 복원점 설정

CATCH

- TRY에서 지정된 signal 일 경우 CATCH 이하 블럭 코드 실행

FINAL

- 처리할 signal의 지정 해제

 

- 어느 함수에서 signal 이 발생했는지 파악하는 것은 여러 모듈을 사용하는 환경에서 필수불가결한 요소이다.

Runtime에 코드 크기를 구하기 위해 위처럼 수행되지 않는 더미 instruction을 삽입하고
Function에서 해당 instruction을 검색하는 방식으로 함수 길이를 runtime에 구할 수 있다.

 

#if defined(__aarch64__) || defined(__arm__)​
#define MARK_FN_END()  {                     \​
    asm volatile("b 1f; .int 0xdeadbeaf;1:");\​
}​
#elif defined(__i386__) || defined(__amd64__)​
#define MARK_FN_END()  {                       \​
    asm volatile("jmp 1f; .int 0xdeadbeaf;1:");\​
}​
#else​
#error "not expected architecture"​
#endif​

​
#define ASSUMED_FN_SIZE 0x150​
unsigned long get_func_size(unsigned long faddr)​
{​
    unsigned long res = 0;​
    unsigned char *fptr = (unsigned char *)faddr;​
    for(int MAX = 100; MAX > 0; fptr++, MAX--)​
    {​
        if (*(unsigned int *)fptr == 0xdeadbeaf) {​
            res = ((unsigned long)fptr) - faddr;​
            break;​
        }​
    }​
​
    return res;​
}​

위는 함수 크기를 구하는 함수이다.

 

 

sigsetjmp/siglongjmp를 통해 복원지점의 설정/복원이 가능하다.

#define TRY(sig, order, func)                \​
    set_signal_handler(order);         \​
    set_watch_func((unsigned long)&func, get_func_size((unsigned long)&func)); \​
    sig = sigsetjmp(point, 1);         \​
    if (sig == 0)​
    
    
    
    static void signal_handler(int sig, siginfo_t *si, void *arg)​
{​
    <...>​

    if (is_watched((unsigned long)mc.gregs[pc_order])) {​
        e.code = sig;​
        e.addr = (unsigned long)mc.gregs[pc_order]; // for test​
        siglongjmp(point, sig);​
    <...>​

 

signal 이 발생해 signal handler 로 진입하게 되면 지정된 함수가 맞는지 확인하고 siglongjmp로 복원한다.

static void signal_handler(int sig siginfo_t *si, void *arg) {​
    ucontext_t *ctx = (ucontext_t *)arg;​
    mcontext_t mc = ctx->uc_mcontext;​
    int pc_order = 0;​
    int sp_order = 0;​
#ifdef __x86_64__​
    pc_order = 16;​
    sp_order = 15;​
#elif if __aarch64__​
    pc_order = 32;​
    sp_order = 31;​
#endif​
    if (is_watched((unsigned long)mc.gregs[pc_order])) {​
        e.code = sig;​
        e.addr = (unsigned long)mc.gregs[pc_order]; // for test​
        siglongjmp(point, sig);​

    <...>​

 

예제코드를 실행하면 다음과 같은 결과를 얻을 수 있다

 

새로운 코드를 반영하고자 할 때, 마음 한켠엔 늘 불안함이 존재한다.
새 코드를 일괄 반영하지 말고 코드가 비정상 동작을 했을때 해당 기능을 일시 비활성화 하고 리포트 하는 과정을 만들면 좀 나아지지 않을까 싶다. 또한 기존 코드를 개선할 때도 일괄 반영하지말고 구 코드를 남겨놓고 새 코드에서 문제가 발생할 경우 구 코드로 일시 대체하고 문제를 해결해나갈 버퍼를 확보할 수 있다면 도움이 될 수 있을 것이라고 생각한다.

 

- 끗 -